diff --git a/AGENTS.md b/AGENTS.md index c2a2889a..48e738d6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,6 +15,13 @@ ArchiveBox is the full self-hosted web archiving app. Keep this repo on the `dev - Trace root causes from observed behavior. Do not paper over failures with retries, wider timeouts, broad fallbacks, or looser assertions. - Read `README.md` for the full setup, CLI, Docker, API, and release surface. +## Concurrency Contract + +- A collection has one orchestrator at a time. Local PID/process checks may warn about obvious same-machine duplicates, but must not claim to enforce ownership across machines or shared filesystems. +- SQLite remains supported with concurrent short writes from CLI and server processes. Keep the SQLite database on a local filesystem, never NFS/SMB, and use short autocommit/CAS updates instead of long transactions or database locks. +- Network calls, hook execution, filesystem migrations, and other long work belong in the orchestrator. Never hold a database transaction or lock across that work. +- PostgreSQL and shared data directories are the path to future multi-machine scheduling. Coordinate that work through database-backed per-crawl/per-snapshot claims; do not introduce file leases or timer-based orchestrator election. + ## Development Setup ```bash diff --git a/Dockerfile b/Dockerfile index 856ec7ed..817dcb42 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,10 +10,10 @@ # --build-context abx-plugins=../abx-plugins \ # -t archivebox/abx-dl:dev # docker buildx build . -f Dockerfile \ -# --build-arg ABX_DL_IMAGE=archivebox/abx-dl:1.12.225 \ +# --build-arg ABX_DL_IMAGE=archivebox/abx-dl:1.12.233 \ # -t archivebox:multistage -ARG ABX_DL_IMAGE=archivebox/abx-dl:1.12.225 +ARG ABX_DL_IMAGE=archivebox/abx-dl:1.12.233 FROM archivebox/sonic:1.4.9 AS sonic FROM ${ABX_DL_IMAGE} AS archivebox-runtime-base diff --git a/archivebox/api/v1_core.py b/archivebox/api/v1_core.py index cc7ec6f6..0ade551d 100644 --- a/archivebox/api/v1_core.py +++ b/archivebox/api/v1_core.py @@ -342,21 +342,10 @@ def _parse_archiveresult_upload_int(value: str, field_name: str, *, default: int def _summarize_archiveresult_output_files(output_files: dict[str, dict[str, Any]]) -> tuple[int, str]: - mime_sizes: dict[str, int] = defaultdict(int) - total_size = 0 - for metadata in output_files.values(): - if not isinstance(metadata, dict): - continue - try: - size = max(int(metadata.get("size") or 0), 0) - except (TypeError, ValueError): - size = 0 - mime_type = str(metadata.get("mimetype") or "").strip() - total_size += size - if mime_type and size: - mime_sizes[mime_type] += size - output_mimetypes = ",".join(mime for mime, _size in sorted(mime_sizes.items(), key=lambda item: item[1], reverse=True)) - return total_size, output_mimetypes + from abx_dl.output_files import OutputManifest + + manifest = OutputManifest.from_value(output_files) + return manifest.total_size, ",".join(manifest.mimetypes) def _get_snapshot_by_ref(snapshot_id: str): diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py index 596cd497..1f5152ae 100644 --- a/archivebox/cli/archivebox_add.py +++ b/archivebox/cli/archivebox_add.py @@ -213,7 +213,7 @@ def add( print(f"[green]\\[+] Created Crawl {crawl.id} with max_depth={depth}[/green]") print(f" [dim]First URL: {first_url}[/dim]") - # 3. The CrawlMachine will create Snapshots from all URLs when started + # 3. The runner will create Snapshots from all URLs after claiming the Crawl # Parser extractors run on snapshots and discover more URLs # Discovered URLs become child Snapshots (depth+1) diff --git a/archivebox/cli/archivebox_extract.py b/archivebox/cli/archivebox_extract.py index 478377f4..f5f915fd 100644 --- a/archivebox/cli/archivebox_extract.py +++ b/archivebox/cli/archivebox_extract.py @@ -296,7 +296,7 @@ def run_plugins( if requested_rows: # Search indexing on a sealed Snapshot is the only targeted hook # allowed to bypass the normal lifecycle. Every other requested - # plugin requeues its Snapshot through the unified state machine. + # plugin requeues its Snapshot through the unified lifecycle. affected_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in rows_to_queue} if preserve_queued and queued_rows: queued_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in queued_rows} diff --git a/archivebox/cli/archivebox_install.py b/archivebox/cli/archivebox_install.py index 47c53bd1..6dfc4796 100755 --- a/archivebox/cli/archivebox_install.py +++ b/archivebox/cli/archivebox_install.py @@ -54,7 +54,7 @@ def _resolve_install_targets( def _install_raw_binary_names(binary_names: list[str], binproviders: str) -> None: - """Install user-requested standalone binaries through the Binary state machine.""" + """Install user-requested standalone binaries through the Binary lifecycle.""" from django.utils import timezone from archivebox.machine.models import Binary, Machine, _canonical_binary_name @@ -179,7 +179,7 @@ def install(binaries: tuple[str, ...] = (), binproviders: str = "*", dry_run: bo run_install(plugin_names=install_plugin_names or None) if raw_binary_names: - print(f"[+] Running direct binary installer via ArchiveBox binary state machine: {', '.join(raw_binary_names)}") + print(f"[+] Running direct binary installer via ArchiveBox binary lifecycle: {', '.join(raw_binary_names)}") print() _install_raw_binary_names(raw_binary_names, binproviders) diff --git a/archivebox/cli/archivebox_pluginmap.py b/archivebox/cli/archivebox_pluginmap.py index 4330e6f9..3bfc2306 100644 --- a/archivebox/cli/archivebox_pluginmap.py +++ b/archivebox/cli/archivebox_pluginmap.py @@ -48,7 +48,7 @@ def pluginmap( Show the current abx-dl event phases and their associated plugin hooks. This command reflects the new bus-driven runtime, not the legacy ArchiveBox - state-machine executor. Event names are normalized to hook prefixes by + event runtime. Event names are normalized to hook prefixes by stripping a trailing `Event`, then ArchiveBox checks whether any matching `on_{EventFamily}__*` scripts actually exist. """ diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index d1c3b310..0b98f5ee 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -81,7 +81,7 @@ def reindex_snapshots( # Search backfill is the one maintenance hook allowed to execute without # reopening a Snapshot. Restrict that exception to already-sealed rows; - # every open lifecycle state remains owned by the normal state machine. + # every open lifecycle state remains owned by the normal runner lifecycle. snapshots = snapshots.filter(status=Snapshot.StatusChoices.SEALED) stats: dict[str, Any] = {"processed": 0, "requested": 0, "queued": 0, "skipped_queued": 0, "reindexed": 0, "snapshot_ids": []} diff --git a/archivebox/config/common.py b/archivebox/config/common.py index 125b71c0..bc83a5fe 100644 --- a/archivebox/config/common.py +++ b/archivebox/config/common.py @@ -16,7 +16,7 @@ from pathlib import Path from typing import Any, ClassVar, cast from urllib.parse import quote, urlparse -from abx_plugins.plugins.base.utils import BASE_CONFIG_PATH, build_config_model, resolve_plugin_configs +from abx_plugins.plugins.base.utils import build_config_model from django.db import DatabaseError from pydantic import BaseModel, Field, PrivateAttr, create_model, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -601,13 +601,9 @@ def _explicit_plugin_enabled_keys(config: Mapping[str, object]) -> set[str]: def _discover_plugin_config_schemas() -> PluginSchemaDocuments: - from archivebox.plugins.discovery import discover_plugin_configs + from archivebox.plugins.discovery import get_plugin_config_resolver - schemas: PluginSchemaDocuments = {} - if BASE_CONFIG_PATH.exists(): - schemas["base"] = json.loads(BASE_CONFIG_PATH.read_text()) - schemas.update(discover_plugin_configs()) - return schemas + return get_plugin_config_resolver().schemas def _plugin_config_properties(plugin_schemas: PluginSchemaDocuments) -> dict[str, dict[str, Any]]: @@ -1195,7 +1191,6 @@ def get_config( explicit_plugin_enabled_keys: set[str] = set() if resolve_plugins: - plugin_schemas = {plugin_name: schema for plugin_name, schema in PLUGIN_CONFIG_SCHEMAS.items() if isinstance(schema, dict)} plugin_global_config = {key: str(value) if isinstance(value, Path) else value for key, value in config_data.items()} crawl_selected_plugins = crawl_config_base and bool(_normalize_plugins_config_value(dict(crawl.config or {}).get("PLUGINS"))) # A frozen crawl-level PLUGINS selector is the exact extractor set for @@ -1219,8 +1214,9 @@ def get_config( **_plugin_user_config(_plugin_input_config(file_config)), **plugin_user_config, } - plugin_sections = resolve_plugin_configs( - plugin_schemas, + from archivebox.plugins.discovery import get_plugin_config_resolver + + plugin_sections = get_plugin_config_resolver().resolve( global_config=plugin_global_config, user_config=plugin_user_config, environ={}, diff --git a/archivebox/core/apps.py b/archivebox/core/apps.py index c2674b6b..0d2f268b 100644 --- a/archivebox/core/apps.py +++ b/archivebox/core/apps.py @@ -10,7 +10,6 @@ class CoreConfig(AppConfig): def ready(self): """Register the archivebox.core.admin_site as the main django admin site""" - import sys from django.utils.autoreload import DJANGO_AUTORELOAD_ENV from archivebox.core.admin_site import register_admin_site @@ -28,11 +27,6 @@ class CoreConfig(AppConfig): pre_save.connect(truncate_overlong_charfields, dispatch_uid="archivebox_truncate_overlong_charfields") - # Import models to register state machines with the registry - # Skip during makemigrations to avoid premature state machine access - if "makemigrations" not in sys.argv: - from archivebox.core import models # noqa: F401 - def _should_prepare_runtime() -> bool: if os.environ.get("ARCHIVEBOX_RUNSERVER") == "1": if os.environ.get("ARCHIVEBOX_AUTORELOAD") == "1": diff --git a/archivebox/core/forms.py b/archivebox/core/forms.py index a9e78c38..8d07d992 100644 --- a/archivebox/core/forms.py +++ b/archivebox/core/forms.py @@ -12,9 +12,9 @@ from archivebox.core.widgets import TagEditorWidget, URLFiltersWidget from archivebox.crawls.schedule_util import validate_schedule from archivebox.misc.util import URL_REGEX, find_all_urls, parse_filesize_to_bytes from archivebox.personas.models import Persona -from archivebox.plugins.discovery import get_plugins +from archivebox.plugins.discovery import get_plugin_catalog from archivebox.plugins.forms import ( - PLUGIN_GROUP_DEFINITIONS, + PLUGIN_GROUPS, TIMEOUT_INPUT_PATTERN, PluginConfigFormMixin, get_choice_field, @@ -333,10 +333,9 @@ class AddLinkForm(PluginConfigFormMixin, forms.Form): if self.can_override_crawl_config: self.build_plugin_groups(get_config(persona=selected_persona) if selected_persona else get_config()) else: - all_plugins = get_plugins() - for field_name, *_rest, plugin_names in PLUGIN_GROUP_DEFINITIONS: - get_choice_field(self, field_name).choices = [(p, p) for p in all_plugins if p in plugin_names] - get_choice_field(self, "other_plugins").choices = [(p, p) for p in all_plugins] + grouped_plugins = get_plugin_catalog().groups() + for category, field_name, _title in PLUGIN_GROUPS: + get_choice_field(self, field_name).choices = [(plugin.name, plugin.name) for plugin in grouped_plugins.get(category, [])] self.plugin_groups = [] def clean(self): diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 33e95b95..7f175359 100644 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -21,7 +21,6 @@ from django.utils import timezone from django.utils.functional import cached_property from django.utils.safestring import mark_safe from django.utils.text import slugify -from statemachine import State, registry from archivebox.base_models.models import ( ModelWithConfig, @@ -54,7 +53,7 @@ from archivebox.plugins.discovery import ( get_plugins, ) from archivebox.uuid_compat import CompactUUIDField, uuid7 -from archivebox.workers.models import ACTIVE_STATE_LEASE_SECONDS, RETRY_AT_MAX, BaseStateMachine, ModelWithStateMachine +from archivebox.workers.models import ACTIVE_STATE_LEASE_SECONDS, RETRY_AT_MAX, ModelWithQueue if TYPE_CHECKING: from archivebox.config.common import ArchiveBoxBaseConfig @@ -536,7 +535,7 @@ class SnapshotManager(models.Manager.from_queryset(SnapshotQuerySet)): # ty: ig return self.get_queryset().delete() -class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithStateMachine): +class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithQueue): BROWSER_EXTENSION_UPLOAD_HOOK_NAME = "on_Snapshot__archivebox_browser_extension_upload" INTERNAL_INPUT_URL = "archivebox://internal" @@ -576,10 +575,10 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW help_text="Current hook step being executed (0-9). Used for sequential hook execution.", ) - retry_at = ModelWithStateMachine.RetryAtField(default=timezone.now) - status = ModelWithStateMachine.StatusField( - choices=ModelWithStateMachine.StatusChoices, - default=ModelWithStateMachine.StatusChoices.QUEUED, + retry_at = ModelWithQueue.RetryAtField(default=timezone.now) + status = ModelWithQueue.StatusField( + choices=ModelWithQueue.StatusChoices, + default=ModelWithQueue.StatusChoices.QUEUED, ) config = models.JSONField(default=dict, null=False, blank=False, editable=True) permissions = models.GeneratedField( @@ -600,10 +599,13 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW tags = models.ManyToManyField(Tag, blank=True, through=SnapshotTag, related_name="snapshot_set", through_fields=("snapshot", "tag")) - state_machine_name = "archivebox.core.models.SnapshotMachine" state_field_name = "status" retry_at_field_name = "retry_at" - StatusChoices = ModelWithStateMachine.StatusChoices + StatusChoices = ModelWithQueue.StatusChoices + INITIAL_STATE = StatusChoices.QUEUED + ACTIVE_STATE = StatusChoices.STARTED + FINAL_STATES = (StatusChoices.SEALED,) + FINAL_OR_ACTIVE_STATES = (*FINAL_STATES, ACTIVE_STATE) active_state = StatusChoices.STARTED delete_after_final_statuses = (StatusChoices.SEALED,) RUNNABLE_STATES = (StatusChoices.QUEUED, StatusChoices.STARTED) @@ -616,11 +618,6 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW objects = SnapshotManager() archiveresult_set: models.Manager["ArchiveResult"] - if TYPE_CHECKING: - - @property - def sm(self) -> "SnapshotMachine": ... - def add_tag_ids(self, tag_ids: Iterable[int | str]) -> None: tag_ids = [tag_id for tag_id in dict.fromkeys(tag_ids) if tag_id] for tag_id in tag_ids: @@ -648,7 +645,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW ModelWithConfig.Meta, ModelWithNotes.Meta, ModelWithHealthStats.Meta, - ModelWithStateMachine.Meta, + ModelWithQueue.Meta, ): app_label = "core" verbose_name = "Snapshot" @@ -769,7 +766,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW Crawl.pause()/cancel() only wake child rows. The runner claims each due Snapshot and lets this method perform the actual child transition, so cancellation stays fast and Snapshot cleanup still runs from the normal - state-machine owner. + lifecycle owner. """ parent_status = Crawl.objects.filter(id=self.crawl_id).values_list("status", flat=True).first() if parent_status == Crawl.StatusChoices.SEALED and self.status != self.StatusChoices.SEALED: @@ -778,7 +775,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW self.refresh_from_db() parent_status = Crawl.objects.filter(id=self.crawl_id).values_list("status", flat=True).first() if parent_status == Crawl.StatusChoices.SEALED and self.status != self.StatusChoices.SEALED: - self.sm.seal() + self.seal() return True if parent_status == Crawl.StatusChoices.PAUSED and self.status not in (self.StatusChoices.PAUSED, self.StatusChoices.SEALED): @@ -834,9 +831,66 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW reset_count += 1 return reset_count, running_count + def start_processing(self) -> bool: + """Atomically move a claimed queued Snapshot into its active lease.""" + owned_retry_at = self.retry_at + now = timezone.now() + lease_until = now + timedelta(seconds=ACTIVE_STATE_LEASE_SECONDS) + updated = ( + type(self) + .objects.filter( + pk=self.pk, + retry_at=owned_retry_at, + status=self.StatusChoices.QUEUED, + ) + .update( + status=self.StatusChoices.STARTED, + retry_at=lease_until, + modified_at=now, + ) + ) + self.refresh_from_db() + return updated == 1 + + def seal(self) -> bool: + """Atomically finalize this Snapshot and reconcile its output metadata.""" + if self.status == self.StatusChoices.SEALED: + return True + now = timezone.now() + updated = ( + type(self) + .objects.filter( + pk=self.pk, + retry_at=self.retry_at, + status__in=self.OPEN_STATES, + ) + .update( + status=self.StatusChoices.SEALED, + retry_at=None, + modified_at=now, + ) + ) + self.refresh_from_db() + if updated == 1: + self.finalize_output_metadata() + return updated == 1 + + def advance_lifecycle(self) -> bool: + """Advance one explicit lifecycle step after the runner claims this row.""" + if self.status == self.StatusChoices.PAUSED: + return False + if self.status == self.StatusChoices.QUEUED: + results = self.archiveresult_set.all() + if results.exists() and not results.exclude(status__in=ArchiveResult.FINAL_STATES).exists(): + return self.seal() + return bool(self.url) and self.start_processing() + if self.status == self.StatusChoices.STARTED and self.is_finished_processing(): + return self.seal() + return False + def cancel(self) -> None: if self.status != self.StatusChoices.SEALED: - self.sm.seal() + self.seal() def get_delete_after_config_value(self): from archivebox.config.common import resolve_delete_after_config_value @@ -943,7 +997,6 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW @property def binary_set(self): """Get all Binary objects used by processes related to this snapshot.""" - from archivebox.machine.models import Binary return Binary.objects.filter(process_set__archiveresult__snapshot_id=self.id).distinct() @@ -2755,11 +2808,11 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW """ return self.create_pending_archiveresults() - def cleanup(self): + def finalize_output_metadata(self) -> None: """ Clean up background ArchiveResult hooks and empty results. - Called by the state machine when entering the 'sealed' state. + Called after entering the sealed state. Reconcile late background outputs and hydrate result metadata. """ # Clean up .pid files from output directory. @@ -2769,11 +2822,10 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW for pid_file in output_dir.glob("**/*.pid"): pid_file.unlink(missing_ok=True) - # Update all background ArchiveResults from filesystem in case - # output arrived late. If there is no snapshot directory, there is - # no filesystem output to reconcile and no reason to hit this query. + # Reconcile late background output without re-running hook-record + # dispatch. The abx-dl event projector is the sole status owner. for ar in self.archiveresult_set.filter(hook_name__contains=".bg."): - ar.update_from_output() + ar.update_output_metadata_from_filesystem(snapshot_dir=output_dir) else: return @@ -3783,162 +3835,6 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None -# ============================================================================= -# Snapshot State Machine -# ============================================================================= - - -class SnapshotMachine(BaseStateMachine): - """ - State machine for managing Snapshot lifecycle. - - Hook Lifecycle: - ┌─────────────────────────────────────────────────────────────┐ - │ QUEUED State │ - │ • Waiting for snapshot to be ready │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when can_start() - ┌─────────────────────────────────────────────────────────────┐ - │ STARTED State → enter_started() │ - │ 1. snapshot.run() │ - │ • discover_hooks('Snapshot') → finds all plugin hooks │ - │ • create_pending_archiveresults() → creates ONE │ - │ ArchiveResult per hook (NO execution yet) │ - │ 2. The shared abx-dl runner executes hooks and the │ - │ projector updates ArchiveResult rows from events │ - │ 3. Advance through steps 0-9 as foreground hooks complete │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when is_finished() - ┌─────────────────────────────────────────────────────────────┐ - │ SEALED State → enter_sealed() │ - │ • cleanup() → kills any background hooks still running │ - │ • Set retry_at=None (no more processing) │ - └─────────────────────────────────────────────────────────────┘ - - https://github.com/ArchiveBox/ArchiveBox/wiki/ArchiveBox-Architecture-Diagrams - """ - - model_attr_name = "snapshot" - - # States - queued = State(value=Snapshot.StatusChoices.QUEUED, initial=True) - started = State(value=Snapshot.StatusChoices.STARTED) - paused = State(value=Snapshot.StatusChoices.PAUSED) - sealed = State(value=Snapshot.StatusChoices.SEALED, final=True) - - # Tick Event (polled by workers) - tick = ( - queued.to(sealed, cond="has_finished_archive_results") - | queued.to.itself(unless="can_start") - | queued.to(started, cond="can_start") - | started.to(sealed, cond="is_finished") - | paused.to.itself() - ) - - # Manual event (can also be triggered by last ArchiveResult finishing) - seal = queued.to(sealed) | started.to(sealed) | paused.to(sealed) - pause_requested = queued.to(paused) | started.to(paused) - resume_requested = paused.to(queued) - - snapshot: Snapshot - - def can_start(self) -> bool: - can_start = bool(self.snapshot.url) - return can_start - - def is_finished(self) -> bool: - """Check if all ArchiveResults for this snapshot are finished.""" - return self.snapshot.is_finished_processing() - - def has_finished_archive_results(self) -> bool: - """A queued snapshot with only final projected rows was interrupted after hook completion.""" - results = self.snapshot.archiveresult_set.all() - return results.exists() and not results.exclude(status__in=ArchiveResult.FINAL_STATES).exists() - - @queued.enter - def enter_queued(self): - self.snapshot.update_and_requeue( - retry_at=timezone.now(), - status=Snapshot.StatusChoices.QUEUED, - ) - - @paused.enter - def enter_paused(self): - self.snapshot.safe_update( - { - "retry_at": RETRY_AT_MAX, - "status": Snapshot.StatusChoices.PAUSED, - }, - extra_filter={"status__in": Snapshot.RUNNABLE_STATES}, - ) - - @started.enter - def enter_started(self): - """Just mark as started. The shared runner creates ArchiveResults and runs hooks.""" - owned_retry_at = self.snapshot.retry_at - now = timezone.now() - lease_until = now + timedelta(seconds=ACTIVE_STATE_LEASE_SECONDS) - # The runner owns queued Snapshot startup through retry_at. Creating - # pending ArchiveResult rows immediately before tick() can touch - # Snapshot.modified_at, so using modified_at CAS here would reject the - # legitimate owner. Keep the write to the scheduler columns only. - updated = Snapshot.objects.filter( - pk=self.snapshot.pk, - retry_at=owned_retry_at, - status=Snapshot.StatusChoices.QUEUED, - ).update( - status=Snapshot.StatusChoices.STARTED, - retry_at=lease_until, - modified_at=now, - ) - if updated != 1: - self.snapshot.refresh_from_db() - return - self.snapshot.status = Snapshot.StatusChoices.STARTED - self.snapshot.retry_at = lease_until - self.snapshot.modified_at = now - - @sealed.enter - def enter_sealed(self): - now = timezone.now() - owned_retry_at = self.snapshot.retry_at - # The runner owns this row via retry_at. Commit the final lifecycle - # state before cleanup so late projectors can update metadata without - # tripping a modified_at CAS while the row still looks QUEUED/STARTED. - updated = ( - type(self.snapshot) - .objects.filter( - pk=self.snapshot.pk, - retry_at=owned_retry_at, - status__in=[ - Snapshot.StatusChoices.QUEUED, - Snapshot.StatusChoices.STARTED, - Snapshot.StatusChoices.PAUSED, - ], - ) - .update( - status=Snapshot.StatusChoices.SEALED, - retry_at=None, - modified_at=now, - ) - ) - if updated != 1: - self.snapshot.refresh_from_db() - return - - self.snapshot.status = Snapshot.StatusChoices.SEALED - self.snapshot.retry_at = None - self.snapshot.modified_at = now - - # Clean up background hooks after the final state is visible in DB. - self.snapshot.cleanup() - - # Crawl finalization is handled by the runner/CrawlService cleanup - # phase. Sealing the parent crawl here races recursive discovery: - # Snapshot hooks can write urls.jsonl just before this state transition, - # and the runner still needs to enqueue those child snapshots. - - class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): class StatusChoices(models.TextChoices): QUEUED = "queued", "Queued" @@ -4448,59 +4344,11 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): self.refresh_from_db() return True - @property - def plugin_module(self) -> Any | None: - # Hook scripts are now used instead of Python plugin modules - # The plugin name maps to hooks in abx_plugins/plugins/{plugin}/ - return None - @staticmethod def _normalize_output_files(raw_output_files: Any) -> dict[str, dict[str, Any]]: - def _enrich_metadata(path: str, metadata: dict[str, Any]) -> dict[str, Any]: - normalized = dict(metadata) - if "extension" not in normalized: - normalized["extension"] = Path(path).suffix.lower().lstrip(".") - if "mimetype" not in normalized: - from abx_dl.output_files import guess_mimetype + from abx_dl.output_files import OutputManifest - guessed = guess_mimetype(path) - if guessed: - normalized["mimetype"] = guessed - return normalized - - if raw_output_files is None: - return {} - if isinstance(raw_output_files, str): - try: - raw_output_files = json.loads(raw_output_files) - except json.JSONDecodeError: - return {} - if isinstance(raw_output_files, dict): - normalized: dict[str, dict[str, Any]] = {} - for path, metadata in raw_output_files.items(): - if not path: - continue - metadata_dict = dict(metadata) if isinstance(metadata, dict) else {} - metadata_dict.pop("path", None) - normalized[str(path)] = _enrich_metadata(str(path), metadata_dict) - return normalized - if isinstance(raw_output_files, (list, tuple, set)): - normalized: dict[str, dict[str, Any]] = {} - for item in raw_output_files: - if isinstance(item, str): - normalized[item] = _enrich_metadata(item, {}) - continue - if not isinstance(item, dict): - continue - path = str(item.get("path") or "").strip() - if not path: - continue - normalized[path] = _enrich_metadata( - path, - {key: value for key, value in item.items() if key != "path" and value not in (None, "")}, - ) - return normalized - return {} + return OutputManifest.from_value(raw_output_files).as_mapping() @staticmethod def _coerce_output_file_size(value: Any) -> int: @@ -4515,16 +4363,8 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): def output_file_paths(self) -> list[str]: return list(self.output_file_map().keys()) - def output_file_count(self) -> int: - return len(self.output_file_paths()) - - def output_size_from_files(self) -> int: - return sum(self._coerce_output_file_size(metadata.get("size")) for metadata in self.output_file_map().values()) - def update_output_metadata_from_filesystem(self, snapshot_dir: Path | None = None, save: bool = True) -> bool: - from collections import defaultdict - - from abx_dl.output_files import guess_mimetype + from abx_dl.output_files import OutputManifest, output_file_from_path if self.plugin == "title": return False @@ -4532,28 +4372,17 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): snapshot_dir = Path(snapshot_dir or self.snapshot.output_dir) exclude_names = {"stdout.log", "stderr.log", "process.pid", "hook.pid", "listener.pid"} output_files: dict[str, dict[str, Any]] = {} - mime_sizes: dict[str, int] = defaultdict(int) - total_size = 0 def add_file(file_path: Path, rel_path: str, *, root_relative: bool = False) -> None: - nonlocal total_size try: if not file_path.is_file() or file_path.name in exclude_names: return - stat = file_path.stat() except OSError: return - mime_type = guess_mimetype(file_path) or "application/octet-stream" - metadata = { - "extension": file_path.suffix.lower().lstrip("."), - "mimetype": mime_type, - "size": stat.st_size, - } + metadata = output_file_from_path(file_path, relative_to=file_path.parent).model_dump(exclude={"path"}) if root_relative: metadata["root_relative"] = True output_files[rel_path] = metadata - mime_sizes[mime_type] += stat.st_size - total_size += stat.st_size for raw_line in str(self.output_str or "").splitlines(): raw_output = raw_line.strip().lstrip("/") @@ -4575,16 +4404,14 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): plugin_dir = snapshot_dir / self.plugin if not output_files and plugin_dir.is_dir(): - for file_path in plugin_dir.rglob("*"): - if not file_path.is_file() or ".hooks" in file_path.parts: - continue - add_file(file_path, str(file_path.relative_to(plugin_dir))) + output_files = OutputManifest.scan(plugin_dir, containment_root=snapshot_dir).as_mapping() if not output_files: return False - sorted_mimes = sorted(mime_sizes.items(), key=lambda item: item[1], reverse=True) - output_mimetypes = ",".join(mime for mime, _ in sorted_mimes) + manifest = OutputManifest.from_value(output_files) + total_size = manifest.total_size + output_mimetypes = ",".join(manifest.mimetypes) if self.output_files == output_files and self.output_size == total_size and self.output_mimetypes == output_mimetypes: return False @@ -4596,9 +4423,6 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): self.save(update_fields=["output_files", "output_size", "output_mimetypes", "modified_at"]) return True - def output_exists(self) -> bool: - return os.path.exists(Path(self.snapshot_dir) / self.plugin) - @staticmethod def _looks_like_output_path(raw_output: str | None, plugin_name: str | None = None) -> bool: value = str(raw_output or "").strip() @@ -4860,207 +4684,6 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): process = self.process_record return process.timeout if process else 120 - def save_search_index(self): - pass - - def update_from_output(self): - """ - Update this ArchiveResult from filesystem logs and output files. - - Used for Snapshot cleanup / orphan recovery when a hook's output exists - on disk but the projector did not finalize the row in the database. - - Updates: - - status, output_str, output_json from ArchiveResult JSONL record - - output_files, output_size, output_mimetypes by walking filesystem - - end_ts, cmd, cmd_version, binary FK - - Processes side-effect records (Snapshot, Tag, etc.) via process_hook_records() - """ - from collections import defaultdict - from pathlib import Path - - from abx_dl.output_files import guess_mimetype - from django.utils import timezone - - from archivebox.machine.models import Process - from archivebox.plugins.hooks import extract_records_from_process, process_hook_records - - plugin_dir = Path(self.pwd) if self.pwd else None - if not plugin_dir or not plugin_dir.exists(): - self.status = self.StatusChoices.FAILED - self.output_str = "Output directory not found" - self.end_ts = timezone.now() - self.save() - return - - records = [] - process = self.process_record - if process: - records = extract_records_from_process(process) - - if not records: - stdout_file = plugin_dir / "stdout.log" - stdout = stdout_file.read_text(errors="replace") if stdout_file.exists() else "" - records = Process.parse_records_from_text(stdout) - - # Find ArchiveResult record and update status/output from it - ar_records = [r for r in records if r.get("type") == "ArchiveResult"] - if ar_records: - hook_data = ar_records[0] - - # Update status - status_map = { - "succeeded": self.StatusChoices.SUCCEEDED, - "failed": self.StatusChoices.FAILED, - "skipped": self.StatusChoices.SKIPPED, - "noresults": self.StatusChoices.NORESULTS, - } - self.status = status_map.get(hook_data.get("status", "failed"), self.StatusChoices.FAILED) - - # Update output fields - self.output_str = hook_data.get("output_str") or hook_data.get("output") or "" - self.output_json = hook_data.get("output_json") - - # Update cmd fields - if hook_data.get("cmd"): - if process: - process.cmd = hook_data["cmd"] - process.save() - self._set_binary_from_cmd(hook_data["cmd"]) - # Note: cmd_version is derived from binary.version, not stored on Process - else: - # No ArchiveResult record: treat background hooks or clean exits as skipped - is_background = False - try: - from archivebox.plugins.hooks import is_background_hook - - is_background = bool(self.hook_name and is_background_hook(self.hook_name)) - except (ImportError, TypeError, ValueError): - is_background = False - - if is_background or (process and process.exit_code == 0): - self.status = self.StatusChoices.SKIPPED - self.output_str = "Hook did not output ArchiveResult record" - else: - self.status = self.StatusChoices.FAILED - self.output_str = "Hook did not output ArchiveResult record" - - # Walk filesystem and populate output_files, output_size, output_mimetypes - exclude_names = {"stdout.log", "stderr.log", "process.pid", "hook.pid", "listener.pid"} - mime_sizes = defaultdict(int) - total_size = 0 - output_files = {} - - for file_path in plugin_dir.rglob("*"): - if not file_path.is_file(): - continue - if ".hooks" in file_path.parts: - continue - if file_path.name in exclude_names: - continue - - try: - stat = file_path.stat() - mime_type = guess_mimetype(file_path) or "application/octet-stream" - - relative_path = str(file_path.relative_to(plugin_dir)) - output_files[relative_path] = { - "extension": file_path.suffix.lower().lstrip("."), - "mimetype": mime_type, - "size": stat.st_size, - } - mime_sizes[mime_type] += stat.st_size - total_size += stat.st_size - except OSError: - continue - - self.output_files = output_files - self.output_size = total_size - sorted_mimes = sorted(mime_sizes.items(), key=lambda x: x[1], reverse=True) - self.output_mimetypes = ",".join(mime for mime, _ in sorted_mimes) - - # Update timestamps - self.end_ts = timezone.now() - - self.save() - - # Process side-effect records (filter Snapshots for depth/URL) - filtered_records = [] - for record in records: - record_type = record.get("type") - - # Skip ArchiveResult records (already processed above) - if record_type == "ArchiveResult": - continue - - # Filter Snapshot records for depth/URL constraints - if record_type == "Snapshot": - url = record.get("url") - if not url: - continue - - depth = record.get("depth", self.snapshot.depth + 1) - if depth > self.snapshot.crawl.max_depth: - continue - - if not self._url_passes_filters(url): - continue - - filtered_records.append(record) - - # Process filtered records with unified dispatcher - overrides = { - "snapshot": self.snapshot, - "crawl": self.snapshot.crawl, - "created_by_id": self.created_by.pk, - } - process_hook_records(filtered_records, overrides=overrides) - - # Cleanup PID files (keep logs even if empty so they can be tailed) - pid_file = plugin_dir / "hook.pid" - pid_file.unlink(missing_ok=True) - - def _set_binary_from_cmd(self, cmd: list) -> None: - """ - Find Binary for command and set binary FK. - - Tries matching by absolute path first, then by binary name. - Only matches binaries on the current machine. - """ - if not cmd: - return - - from archivebox.machine.models import Machine - - bin_path_or_name = cmd[0] if isinstance(cmd, list) else cmd - machine = Machine.current() - - # Try matching by absolute path first - binary = Binary.objects.filter( - abspath=bin_path_or_name, - machine=machine, - ).first() - - if binary: - process = self.process_record - if process: - process.binary = binary - process.save() - return - - # Fallback: match by binary name - bin_name = Path(bin_path_or_name).name - binary = Binary.objects.filter( - name=bin_name, - machine=machine, - ).first() - - if binary: - process = self.process_record - if process: - process.binary = binary - process.save() - def _url_passes_filters(self, url: str) -> bool: """Check if URL passes URL_ALLOWLIST and URL_DENYLIST config filters. @@ -5073,12 +4696,3 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): def output_dir(self) -> Path: """Get the output directory for this plugin's results.""" return Path(self.snapshot.output_dir) / self.plugin - - -# ============================================================================= -# State Machine Registration -# ============================================================================= - -# Manually register state machines with python-statemachine registry -# (normally auto-discovered from statemachines.py, but we define them here for clarity) -registry.register(SnapshotMachine) diff --git a/archivebox/core/recovery_util.py b/archivebox/core/recovery_util.py index 3403ead5..5a3f1266 100644 --- a/archivebox/core/recovery_util.py +++ b/archivebox/core/recovery_util.py @@ -18,7 +18,7 @@ def _canonical_hook_name(hook_name: str) -> str: def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str | None = None) -> dict[str, int]: from archivebox.crawls.models import Crawl from archivebox.core.models import ArchiveResult, Snapshot - from archivebox.services.archive_result_service import _collect_output_metadata + from abx_dl.output_files import OutputManifest from archivebox.machine.models import Process from django.core.exceptions import ValidationError from django.db.models import Exists, OuterRef, Q, Subquery, Value @@ -215,7 +215,10 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str | # A runner can die after the hook Process exits but before the # ProcessCompletedEvent projector links/finalizes ArchiveResult. # Reconstruct only that exact hook row from the durable Process row. - output_files, output_size, output_mimetypes = _collect_output_metadata(plugin_dir) + manifest = OutputManifest.scan(plugin_dir, containment_root=snapshot.output_dir) + output_files = manifest.as_mapping() + output_size = manifest.total_size + output_mimetypes = ",".join(manifest.mimetypes) emitted_records = [ record for record in Process.parse_records_from_text(process.stdout or "") @@ -301,7 +304,7 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str | # Broken lock repair: STARTED + retry_at=NULL is an orphaned ownership # lease. Recovery only unlocks scheduling; the runner owns any subsequent - # state-machine transition, including sealing rows whose children/results + # lifecycle transition, including sealing rows whose children/results # are already final. recoverable_started_crawls = Crawl.objects.filter(status=Crawl.StatusChoices.STARTED).filter( Q(retry_at__isnull=True) | Q(retry_at__gt=now), diff --git a/archivebox/core/templatetags/core_tags.py b/archivebox/core/templatetags/core_tags.py index 7bc5b4d2..89cb1265 100644 --- a/archivebox/core/templatetags/core_tags.py +++ b/archivebox/core/templatetags/core_tags.py @@ -68,14 +68,9 @@ _MEDIA_FILE_EXTS = { def _normalize_output_files(output_files: Any) -> dict[str, dict[str, Any]]: - if isinstance(output_files, dict): - normalized: dict[str, dict[str, Any]] = {} - for path, metadata in output_files.items(): - if not path: - continue - normalized[str(path)] = dict(metadata) if isinstance(metadata, dict) else {} - return normalized - return {} + from abx_dl.output_files import OutputManifest + + return OutputManifest.from_value(output_files).as_mapping() def _snapshot_id(value: Any) -> Any: diff --git a/archivebox/crawls/apps.py b/archivebox/crawls/apps.py index b9e5ed66..ee81278c 100644 --- a/archivebox/crawls/apps.py +++ b/archivebox/crawls/apps.py @@ -5,11 +5,3 @@ class CrawlsConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" name = "archivebox.crawls" label = "crawls" - - def ready(self): - """Import models to register state machines with the registry""" - import sys - - # Skip during makemigrations to avoid premature state machine access - if "makemigrations" not in sys.argv: - from archivebox.crawls.models import CrawlMachine # noqa: F401 diff --git a/archivebox/crawls/models.py b/archivebox/crawls/models.py index a1af3793..c8a46841 100755 --- a/archivebox/crawls/models.py +++ b/archivebox/crawls/models.py @@ -19,7 +19,6 @@ from django.core.validators import MaxValueValidator, MinValueValidator from django.conf import settings from django.urls import reverse_lazy from django.utils import timezone -from statemachine import State, registry from archivebox.config.common import rprint as print from archivebox.core.permissions import PERMISSIONS_VALUES, normalize_permissions @@ -32,7 +31,7 @@ from archivebox.base_models.models import ( ModelWithHealthStats, get_or_create_system_user_pk, ) -from archivebox.workers.models import RETRY_AT_MAX, ModelWithStateMachine, BaseStateMachine +from archivebox.workers.models import ModelWithQueue from archivebox.crawls.schedule_util import next_run_for_schedule, validate_schedule from archivebox.misc.util import parse_date, sanitize_html_text, validate_url, validate_url_length @@ -129,7 +128,7 @@ class CrawlSchedule(ModelWithUUID, ModelWithNotes): ) -class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithStateMachine): +class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): id = CompactUUIDField(primary_key=True, default=uuid7, editable=False, unique=True) created_at = models.DateTimeField(default=timezone.now, db_index=True) created_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, default=get_or_create_system_user_pk, null=False) @@ -158,16 +157,19 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith notes = models.TextField(blank=True, null=False, default="") schedule = models.ForeignKey(CrawlSchedule, on_delete=models.SET_NULL, null=True, blank=True, editable=True) - status = ModelWithStateMachine.StatusField( - choices=ModelWithStateMachine.StatusChoices, - default=ModelWithStateMachine.StatusChoices.QUEUED, + status = ModelWithQueue.StatusField( + choices=ModelWithQueue.StatusChoices, + default=ModelWithQueue.StatusChoices.QUEUED, ) - retry_at = ModelWithStateMachine.RetryAtField(default=timezone.now) + retry_at = ModelWithQueue.RetryAtField(default=timezone.now) - state_machine_name = "archivebox.crawls.models.CrawlMachine" retry_at_field_name = "retry_at" state_field_name = "status" - StatusChoices = ModelWithStateMachine.StatusChoices + StatusChoices = ModelWithQueue.StatusChoices + INITIAL_STATE = StatusChoices.QUEUED + ACTIVE_STATE = StatusChoices.STARTED + FINAL_STATES = (StatusChoices.SEALED,) + FINAL_OR_ACTIVE_STATES = (*FINAL_STATES, ACTIVE_STATE) active_state = StatusChoices.STARTED delete_after_final_statuses = (StatusChoices.SEALED,) RUNNABLE_STATES = (StatusChoices.QUEUED, StatusChoices.STARTED) @@ -177,17 +179,12 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith snapshot_set: models.Manager["Snapshot"] - if TYPE_CHECKING: - - @property - def sm(self) -> "CrawlMachine": ... - class Meta( ModelWithDeleteAfter.Meta, ModelWithOutputDir.Meta, ModelWithConfig.Meta, ModelWithHealthStats.Meta, - ModelWithStateMachine.Meta, + ModelWithQueue.Meta, ): app_label = "crawls" verbose_name = "Crawl" @@ -255,7 +252,7 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith now = timezone.now() # Cancellation seals the Crawl first, then lets the runner seal each - # child Snapshot through its own state machine. Active children that + # child Snapshot through its own lifecycle. Active children that # are already due need no write; the runner will claim them as-is. active_children = self.snapshot_set.filter( status__in=Snapshot.OPEN_STATES, @@ -1296,170 +1293,6 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith return created_snapshots - def install_declared_binaries(self, binary_names: set[str], machine=None) -> None: - """Install crawl-declared binaries through their unified state machine.""" - from archivebox.crawls.locks import binary_lifecycle_lock - from archivebox.machine.models import Binary, Machine - - if not binary_names: - return - - machine = machine or Machine.current() - binaries = Binary.objects.filter(machine=machine, name__in=binary_names).order_by("name") - for binary in binaries: - with binary_lifecycle_lock(str(binary.id)): - binary.refresh_from_db() - if binary.status == Binary.StatusChoices.INSTALLED: - continue - binary.update_and_requeue(retry_at=timezone.now()) - binary.refresh_from_db() - binary.tick_claimed(lock_seconds=600) - - unresolved_binaries = list( - Binary.objects.filter( - machine=machine, - name__in=binary_names, - ) - .exclude( - status=Binary.StatusChoices.INSTALLED, - ) - .order_by("name"), - ) - if unresolved_binaries: - binary_details = ", ".join( - f"{binary.name} (status={binary.status}, retry_at={binary.retry_at})" for binary in unresolved_binaries - ) - raise RuntimeError( - f"Crawl dependencies failed to install before continuing: {binary_details}", - ) - - def run(self) -> "Snapshot | None": - """ - Execute this Crawl: run hooks, process JSONL, create snapshots. - - Called by the state machine when entering the 'started' state. - - Returns: - The root Snapshot for this crawl, or None for system crawls that don't create snapshots - """ - import time - from archivebox.plugins.hooks import run_hook, discover_hooks, process_hook_records - from archivebox.config.common import get_config - from archivebox.machine.models import Machine - - def get_runtime_config(): - return get_config(crawl=self).for_crawl_runtime( - crawl=self, - persona=persona, - runtime_overrides=persona_runtime_overrides, - ) - - system_task = self.get_system_task() - if system_task == "archivebox://update": - from archivebox.cli.archivebox_update import process_all_db_snapshots - - process_all_db_snapshots() - return None - - machine = Machine.current() - declared_binary_names: set[str] = set() - persona_runtime_overrides: dict[str, str] = {} - persona = self.resolve_persona() - if persona: - base_runtime_config = get_config(crawl=self, persona=persona) - chrome_binary = str(base_runtime_config.get("CHROME_BINARY") or "") - persona_runtime_overrides = persona.prepare_runtime_for_crawl( - crawl=self, - chrome_binary=chrome_binary, - ) - - def run_crawl_hook(hook: Path) -> set[str]: - primary_url = next( - (line.strip() for line in self.urls.splitlines() if line.strip()), - self.urls.strip(), - ) - - hook_start = time.time() - plugin_name = hook.parent.name - output_dir = self.output_dir / plugin_name - output_dir.mkdir(parents=True, exist_ok=True) - - process = run_hook( - hook, - output_dir=output_dir, - config=get_runtime_config(), - crawl_id=str(self.id), - source_url=self.urls, - url=primary_url, - snapshot_id=str(self.id), - ) - hook_elapsed = time.time() - hook_start - if hook_elapsed > 0.5: - print(f"[yellow]⏱️ Hook {hook.name} took {hook_elapsed:.2f}s[/yellow]") - - if process.status == process.StatusChoices.RUNNING: - if process.poll() is None: - return set() - - from archivebox.plugins.hooks import extract_records_from_process - - records = [] - # A hook can exit before its completed Process metadata is visible. - # Give successful hooks a brief chance to flush JSONL stdout into - # the Process row before downstream hooks. - for delay in (0.0, 0.05, 0.1, 0.25, 0.5): - if delay: - time.sleep(delay) - records = extract_records_from_process(process) - if records: - break - if records: - print(f"[cyan]📝 Processing {len(records)} records from {hook.name}[/cyan]") - for record in records[:3]: - print(f" Record: type={record.get('type')}, keys={list(record.keys())[:5]}") - if system_task: - records = [record for record in records if record.get("type") in ("BinaryRequest", "Binary")] - overrides = {"crawl": self} - stats = process_hook_records(records, overrides=overrides) - if stats: - print(f"[green]✓ Created: {stats}[/green]") - - hook_binary_names = { - str(record.get("name")).strip() - for record in records - if record.get("type") in ("BinaryRequest", "Binary") and record.get("name") - } - hook_binary_names.discard("") - if hook_binary_names: - declared_binary_names.update(hook_binary_names) - return hook_binary_names - - hooks = discover_hooks("Crawl", config=get_runtime_config()) - - for hook in hooks: - hook_binary_names = run_crawl_hook(hook) - if hook_binary_names: - self.install_declared_binaries(hook_binary_names, machine=machine) - - # Safety check: don't create snapshots if any crawl-declared dependency - # is still unresolved after all crawl hooks have run. - self.install_declared_binaries(declared_binary_names, machine=machine) - - # Create snapshots from all URLs in self.urls - if system_task: - leaked_snapshots = self.snapshot_set.all() - if leaked_snapshots.exists(): - leaked_count = leaked_snapshots.count() - leaked_snapshots.delete() - print(f"[yellow]⚠️ Removed {leaked_count} leaked snapshot(s) created during system crawl {system_task}[/yellow]") - return None - - self.create_snapshots_from_urls() - - # Return first snapshot for this crawl (newly created or existing) - # This ensures the crawl doesn't seal if snapshots exist, even if they weren't just created - return self.snapshot_set.first() - def is_finished(self) -> bool: """Check if crawl is finished (all snapshots sealed or no snapshots exist).""" from archivebox.core.models import Snapshot @@ -1483,11 +1316,64 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith return True - def cleanup(self): - """Clean up background hooks and run on_CrawlEnd hooks.""" - from archivebox.plugins.hooks import run_hook, discover_hooks + def can_start(self) -> bool: + return bool(self.urls and self.get_urls_list()) - # Clean up .pid files from output directory + def has_finished_snapshots(self) -> bool: + from archivebox.core.models import Snapshot + + snapshots = self.snapshot_set.all() + return snapshots.exists() and not snapshots.exclude(status=Snapshot.StatusChoices.SEALED).exists() + + def mark_started(self) -> bool: + now = timezone.now() + updated = self.safe_update( + { + "status": self.StatusChoices.STARTED, + "retry_at": now + timedelta(seconds=2), + }, + extra_filter={"status": self.StatusChoices.QUEUED}, + ) + return updated + + def seal(self) -> bool: + """Finalize a runner-owned Crawl without dispatching hooks directly.""" + now = timezone.now() + updated = self.safe_update( + { + "status": self.StatusChoices.SEALED, + "retry_at": None, + "modified_at": now, + }, + refresh=False, + extra_filter={"status__in": (*self.RUNNABLE_STATES, self.StatusChoices.SEALED)}, + ) + if not updated: + self.refresh_from_db() + return False + self.status = self.StatusChoices.SEALED + self.retry_at = None + self.modified_at = now + self.schedule_child_snapshots_for_sealing() + self.cleanup_runtime() + return True + + def advance_lifecycle(self) -> bool: + """Advance one explicit lifecycle step after the runner claims this row.""" + if self.status == self.StatusChoices.PAUSED: + return False + if self.status == self.StatusChoices.QUEUED: + if self.has_finished_snapshots(): + return self.seal() + if not self.can_start(): + return False + return self.mark_started() + if self.status == self.StatusChoices.STARTED and self.is_finished(): + return self.seal() + return False + + def cleanup_runtime(self) -> None: + """Remove runner-owned runtime artifacts after abx-dl cleanup hooks finish.""" if self.output_dir.exists(): for pid_file in self.output_dir.glob("**/*.pid"): pid_file.unlink(missing_ok=True) @@ -1495,207 +1381,3 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith persona = self.resolve_persona() if persona: persona.cleanup_runtime_for_crawl(self) - - # Run on_CrawlEnd hooks - from archivebox.config.common import get_config - - config = get_config(crawl=self) - - hooks = discover_hooks("CrawlEnd", config=config) - - for hook in hooks: - plugin_name = hook.parent.name - output_dir = self.output_dir / plugin_name - output_dir.mkdir(parents=True, exist_ok=True) - - process = run_hook( - hook, - output_dir=output_dir, - config=config, - crawl_id=str(self.id), - source_url=self.urls, # Pass full newline-separated URLs - ) - - # Log failures but don't block - if process.exit_code != 0: - print(f"[yellow]⚠️ CrawlEnd hook failed: {hook.name}[/yellow]") - - -# ============================================================================= -# State Machines -# ============================================================================= - - -class CrawlMachine(BaseStateMachine): - crawl: Crawl - - """ - State machine for managing Crawl lifecycle. - - Hook Lifecycle: - ┌─────────────────────────────────────────────────────────────┐ - │ QUEUED State │ - │ • Waiting for crawl to be ready (has URLs) │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when can_start() - ┌─────────────────────────────────────────────────────────────┐ - │ STARTED State → enter_started() │ - │ 1. crawl.run() │ - │ • discover_hooks('Crawl') → finds all crawl hooks │ - │ • For each hook: │ - │ - run_hook(script, output_dir, ...) │ - │ - Parse JSONL from hook output │ - │ - process_hook_records() → creates Snapshots │ - │ • create_snapshots_from_urls() → from self.urls field │ - │ │ - │ 2. Snapshots process independently with their own │ - │ state machines (see SnapshotMachine) │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when is_finished() - ┌─────────────────────────────────────────────────────────────┐ - │ SEALED State → enter_sealed() │ - │ • cleanup() → runs on_CrawlEnd hooks, kills background │ - │ • Set retry_at=None (no more processing) │ - └─────────────────────────────────────────────────────────────┘ - """ - - model_attr_name = "crawl" - - # States - queued = State(value=Crawl.StatusChoices.QUEUED, initial=True) - started = State(value=Crawl.StatusChoices.STARTED) - paused = State(value=Crawl.StatusChoices.PAUSED) - sealed = State(value=Crawl.StatusChoices.SEALED, final=True) - - # Tick Event (polled by workers) - tick = ( - queued.to(sealed, cond="has_finished_snapshots") - | queued.to.itself(unless="can_start") - | queued.to(started, cond="can_start") - | started.to(sealed, cond="is_finished") - | paused.to.itself() - ) - - # Manual event (triggered by last Snapshot sealing, or by direct - # index-only/bg creation when every requested URL is rejected before any - # Snapshot rows exist). - seal = queued.to(sealed) | started.to(sealed) | paused.to(sealed) - pause_requested = queued.to(paused) | started.to(paused) - resume_requested = paused.to(queued) - - def can_start(self) -> bool: - if not self.crawl.urls: - print(f"[red]⚠️ Crawl {self.crawl.id} cannot start: no URLs[/red]") - return False - urls_list = self.crawl.get_urls_list() - if not urls_list: - print(f"[red]⚠️ Crawl {self.crawl.id} cannot start: no valid URLs in urls field[/red]") - return False - return True - - def is_finished(self) -> bool: - """Check if all Snapshots for this crawl are finished.""" - return self.crawl.is_finished() - - def has_finished_snapshots(self) -> bool: - """A queued crawl with only final Snapshot rows was interrupted before sealing.""" - from archivebox.core.models import Snapshot - - snapshots = self.crawl.snapshot_set.all() - return snapshots.exists() and not snapshots.exclude(status=Snapshot.StatusChoices.SEALED).exists() - - @queued.enter - def enter_queued(self): - self.crawl.update_and_requeue( - retry_at=timezone.now(), - status=Crawl.StatusChoices.QUEUED, - ) - - @started.enter - def enter_started(self): - import sys - - print(f"[cyan]🔄 CrawlMachine.enter_started() - creating snapshots for {self.crawl.id}[/cyan]", file=sys.stderr) - - try: - # Run the crawl - runs hooks, processes JSONL, creates snapshots - first_snapshot = self.crawl.run() - - if first_snapshot: - print( - f"[cyan]🔄 Created {self.crawl.snapshot_set.count()} snapshot(s), first: {first_snapshot.url}[/cyan]", - file=sys.stderr, - ) - # Update status to STARTED - # Set retry_at to near future so tick() can poll and check is_finished() - self.crawl.update_and_requeue( - retry_at=timezone.now() + timedelta(seconds=2), - status=Crawl.StatusChoices.STARTED, - ) - else: - # No snapshots (system crawl that only runs setup hooks) - print("[cyan]🔄 No snapshots created, sealing crawl immediately[/cyan]", file=sys.stderr) - # Seal immediately since there's no work to do - self.seal() - - except Exception as e: - print(f"[red]⚠️ Crawl {self.crawl.id} failed to start: {e}[/red]") - import traceback - - traceback.print_exc() - raise - - @paused.enter - def enter_paused(self): - paused = self.crawl.safe_update( - { - "retry_at": RETRY_AT_MAX, - "status": Crawl.StatusChoices.PAUSED, - }, - extra_filter={"status__in": Crawl.RUNNABLE_STATES}, - ) - if paused: - self.crawl.schedule_child_snapshots_for_pause() - - @sealed.enter - def enter_sealed(self): - now = timezone.now() - self.crawl.status = Crawl.StatusChoices.SEALED - self.crawl.retry_at = None - # Guard: never seal a row that a concurrent writer flipped to PAUSED. - # Sealing is idempotent (SEALED→SEALED is a no-op rewrite), so - # status__in covers both the QUEUED/STARTED→SEALED transition and the - # rare re-entry case. - updated = self.crawl.safe_update( - { - "status": Crawl.StatusChoices.SEALED, - "retry_at": None, - "modified_at": now, - }, - refresh=False, - extra_filter={ - "status__in": [ - Crawl.StatusChoices.QUEUED, - Crawl.StatusChoices.STARTED, - Crawl.StatusChoices.SEALED, - ], - }, - ) - if not updated: - self.crawl.refresh_from_db() - return - self.crawl.modified_at = now - - self.crawl.schedule_child_snapshots_for_sealing() - # Clean up background hooks and run on_CrawlEnd hooks after the final - # state is visible so cleanup projectors cannot resurrect the crawl. - self.crawl.cleanup() - - -# ============================================================================= -# Register State Machines -# ============================================================================= - -# Manually register state machines with python-statemachine registry -# (normally auto-discovered from statemachines.py, but we define them here for clarity) -registry.register(CrawlMachine) diff --git a/archivebox/machine/apps.py b/archivebox/machine/apps.py index f4834e4c..442c3db5 100644 --- a/archivebox/machine/apps.py +++ b/archivebox/machine/apps.py @@ -10,14 +10,6 @@ class MachineConfig(AppConfig): label = "machine" # Explicit label for migrations verbose_name = "Machine Info" - def ready(self): - """Import models to register state machines with the registry""" - import sys - - # Skip during makemigrations to avoid premature state machine access - if "makemigrations" not in sys.argv: - from archivebox.machine import models # noqa: F401 - def register_admin(admin_site): from archivebox.machine.admin import register_admin diff --git a/archivebox/machine/models.py b/archivebox/machine/models.py index b90015db..07ba0030 100755 --- a/archivebox/machine/models.py +++ b/archivebox/machine/models.py @@ -12,8 +12,6 @@ from archivebox.uuid_compat import CompactUUIDField, uuid7 from datetime import timedelta, datetime from typing import TYPE_CHECKING, Any, cast -from statemachine import State, registry - from django.db import IntegrityError, transaction from django.db import models from django.db.models import Q, QuerySet @@ -23,7 +21,7 @@ from django.utils.functional import cached_property from archivebox.config import CONSTANTS from archivebox.config.common import rprint from archivebox.base_models.models import ModelWithDeleteAfter, ModelWithHealthStats, normalize_config_json_values -from archivebox.workers.models import BaseStateMachine, ModelWithStateMachine +from archivebox.workers.models import ModelWithQueue from .detect import get_host_guid, get_os_info, get_vm_info, get_host_network, get_host_stats _psutil: Any | None = None @@ -515,19 +513,21 @@ class BinaryManager(models.Manager): ) -class Binary(ModelWithHealthStats, ModelWithStateMachine): +class Binary(ModelWithHealthStats, ModelWithQueue): """ Tracks a binary on a specific machine. - Simple state machine with 2 states: + Simple queue lifecycle with 2 states: - queued: Binary needs to be installed - installed: Binary installed successfully (abspath, version, sha256 populated) Installation is synchronous during queued→installed transition. If installation fails, Binary stays in queued with retry_at set for later retry. - State machine calls run(), which emits an abxpkg BinaryRequestEvent through - the ArchiveBox runner and installs the binary using the specified providers. + BinaryService claims queued rows with a conditional update, then run() + emits an abxpkg BinaryRequestEvent and persists the resolved installation. + The database row is the only lifecycle state; there is intentionally no + second in-memory state machine to reconcile after a worker interruption. """ class StatusChoices(models.TextChoices): @@ -566,9 +566,9 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): version = models.CharField(max_length=32, default="", null=False, blank=True) sha256 = models.CharField(max_length=64, default="", null=False, blank=True) - # State machine fields - status = ModelWithStateMachine.StatusField(choices=StatusChoices.choices, default=StatusChoices.QUEUED, max_length=16) - retry_at = ModelWithStateMachine.RetryAtField( + # Durable queue lifecycle fields + status = ModelWithQueue.StatusField(choices=StatusChoices.choices, default=StatusChoices.QUEUED, max_length=16) + retry_at = ModelWithQueue.RetryAtField( default=timezone.now, help_text="When to retry this binary installation", ) @@ -579,18 +579,16 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): machine_id: uuid.UUID - state_machine_name: str | None = "archivebox.machine.models.BinaryMachine" + INITIAL_STATE = StatusChoices.QUEUED + ACTIVE_STATE = StatusChoices.QUEUED + FINAL_STATES = (StatusChoices.INSTALLED,) + FINAL_OR_ACTIVE_STATES = (*FINAL_STATES, ACTIVE_STATE) active_state: str = StatusChoices.QUEUED warn_on_save_outside_runner = False objects = BinaryManager() # pyright: ignore[reportIncompatibleVariableOverride] - if TYPE_CHECKING: - - @property - def sm(self) -> BinaryMachine: ... - - class Meta(ModelWithHealthStats.Meta, ModelWithStateMachine.Meta): + class Meta(ModelWithHealthStats.Meta, ModelWithQueue.Meta): app_label = "machine" verbose_name = "Binary" verbose_name_plural = "Binaries" @@ -604,6 +602,10 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): """A binary is valid if it has a resolved path and is marked installed.""" return bool(self.abspath) and self.status == self.StatusChoices.INSTALLED + @property + def can_install(self) -> bool: + return bool(self.name and self.binproviders) + @cached_property def binary_info(self) -> dict: """Return info about the binary.""" @@ -748,11 +750,43 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): run_binary(str(self.id)) + def install(self) -> bool: + """Run one synchronous installation attempt for a claimed Binary.""" + if self.status == self.StatusChoices.INSTALLED: + return True + if not self.can_install: + return False + + rprint(f"[cyan] 🔄 installing {self.name}[/cyan]", file=sys.stderr) + self.run() + self.refresh_from_db() + if self.status != self.StatusChoices.INSTALLED: + self.update_and_requeue( + retry_at=timezone.now() + timedelta(seconds=300), + status=self.StatusChoices.QUEUED, + ) + self.increment_health_stats(success=False) + raise RuntimeError(f"Binary {self.name} installation failed") + + self.update_and_requeue(retry_at=None, status=self.StatusChoices.INSTALLED) + self.increment_health_stats(success=True) + return True + + def advance_lifecycle(self) -> bool: + """Advance the explicit binary lifecycle after its queue row is claimed.""" + return self.install() + + def install_claimed(self, *, lock_seconds: int = 600) -> bool: + if not self.claim_processing_lock(lock_seconds=lock_seconds): + return False + self.refresh_from_db() + return self.advance_lifecycle() + def cleanup(self): """ Clean up background binary installation hooks. - Called by state machine if needed (not typically used for binaries + Called after an installation attempt if needed (not typically used for binaries since installations are foreground, but included for consistency). """ @@ -953,12 +987,14 @@ class Process(ModelWithDeleteAfter, models.Model): One Process can optionally be associated with an ArchiveResult (via OneToOne), but Process can also exist standalone for internal operations. - Follows the unified state machine pattern: + Follows the unified process lifecycle: - queued: Process ready to launch - running: Process actively executing - exited: Process completed (check exit_code for success/failure) - State machine calls launch() to spawn the process and monitors its lifecycle. + Direct Process methods and the abx-dl event projector own these transitions. + Keeping the DB row as the only lifecycle state makes interrupted subprocess + recovery observable without reconciling a second in-memory state machine. """ class StatusChoices(models.TextChoices): @@ -1117,7 +1153,7 @@ class Process(ModelWithDeleteAfter, models.Model): # Reverse relation to ArchiveResult (OneToOne from AR side) # archiveresult: OneToOneField defined on ArchiveResult model - # State machine fields + # Durable process lifecycle fields status = models.CharField( max_length=16, choices=StatusChoices.choices, @@ -1138,7 +1174,6 @@ class Process(ModelWithDeleteAfter, models.Model): children: models.Manager[Process] archiveresult: ArchiveResult - state_machine_name: str = "archivebox.machine.models.ProcessMachine" delete_after_final_statuses = (StatusChoices.EXITED,) objects = ProcessManager() # pyright: ignore[reportIncompatibleVariableOverride] @@ -1308,8 +1343,8 @@ class Process(ModelWithDeleteAfter, models.Model): """ Compare-and-swap update for short Process scheduler writes. - Process is not a ModelWithStateMachine subclass yet, but its - state-machine methods still need the same modified_at CAS behavior as + Process is not a ModelWithQueue subclass, but its scheduler methods + still need the same modified_at CAS behavior as Crawl/Snapshot/Binary without falling back to save(). """ values = dict(update_fields) @@ -2233,8 +2268,8 @@ class Process(ModelWithDeleteAfter, models.Model): """ Gracefully terminate process: SIGTERM → wait → SIGKILL. - This consolidates the scattered SIGTERM/SIGKILL logic from: - - crawls/models.py Crawl.cleanup() + This consolidates SIGTERM/SIGKILL logic used by: + - workers/management/commands/runner_watch.py - workers/pid_utils.py stop_worker() - supervisord_util.py stop_existing_supervisord_process() @@ -2297,8 +2332,8 @@ class Process(ModelWithDeleteAfter, models.Model): Uses parallel polling approach - sends SIGTERM to all processes at once, then polls all simultaneously with individual deadline tracking. - This consolidates the scattered child-killing logic from: - - crawls/models.py Crawl.cleanup() os.killpg() + This consolidates child-killing logic used by: + - core/takeover_util.py - supervisord_util.py stop_existing_supervisord_process() Args: @@ -2598,206 +2633,3 @@ class Process(ModelWithDeleteAfter, models.Model): if cleaned: rprint(f"[yellow]🧹 Cleaned up {cleaned} orphaned worker/hook process record(s)[/yellow]") return cleaned - - -# ============================================================================= -# Binary State Machine -# ============================================================================= - - -class BinaryMachine(BaseStateMachine): - """ - State machine for managing Binary installation lifecycle. - - Simple 2-state machine: - ┌─────────────────────────────────────────────────────────────┐ - │ QUEUED State │ - │ • Binary needs to be installed │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when can_install() - ↓ Synchronous installation during transition - ┌─────────────────────────────────────────────────────────────┐ - │ INSTALLED State │ - │ • Binary installed (abspath, version, sha256 set) │ - │ • Health stats incremented │ - └─────────────────────────────────────────────────────────────┘ - - If installation fails, Binary stays in QUEUED with retry_at bumped. - """ - - model_attr_name = "binary" - binary: Binary - - # States - queued = State(value=Binary.StatusChoices.QUEUED, initial=True) - installed = State(value=Binary.StatusChoices.INSTALLED, final=True) - - # Tick Event - install happens during transition - tick = queued.to.itself(unless="can_install") | queued.to(installed, cond="can_install", on="on_install") - - def can_install(self) -> bool: - """Check if binary installation can start.""" - return bool(self.binary.name and self.binary.binproviders) - - @queued.enter - def enter_queued(self): - """Binary is queued for installation.""" - self.binary.update_and_requeue( - retry_at=timezone.now(), - status=Binary.StatusChoices.QUEUED, - ) - - def on_install(self): - """Called during queued→installed transition. Runs installation synchronously.""" - import sys - - rprint(f"[cyan] 🔄 BinaryMachine.on_install() - installing {self.binary.name}[/cyan]", file=sys.stderr) - - # Run installation hooks (synchronous, updates abspath/version/sha256 and sets status) - self.binary.run() - - # Check if installation succeeded by looking at updated status - # Note: Binary.run() updates self.binary.status internally but doesn't refresh our reference - self.binary.refresh_from_db() - - if self.binary.status != Binary.StatusChoices.INSTALLED: - # Installation failed - abort transition, stay in queued - rprint(f"[red] ❌ BinaryMachine - {self.binary.name} installation failed, retrying later[/red]", file=sys.stderr) - - # Bump retry_at to try again later - self.binary.update_and_requeue( - retry_at=timezone.now() + timedelta(seconds=300), # Retry in 5 minutes - status=Binary.StatusChoices.QUEUED, # Ensure we stay queued - ) - - # Increment health stats for failure - self.binary.increment_health_stats(success=False) - - # Abort the transition - this will raise an exception and keep us in queued - raise Exception(f"Binary {self.binary.name} installation failed") - - rprint(f"[cyan] ✅ BinaryMachine - {self.binary.name} installed successfully[/cyan]", file=sys.stderr) - - @installed.enter - def enter_installed(self): - """Binary installed successfully.""" - self.binary.update_and_requeue( - retry_at=None, - status=Binary.StatusChoices.INSTALLED, - ) - - # Increment health stats - self.binary.increment_health_stats(success=True) - - -# ============================================================================= -# Process State Machine -# ============================================================================= - - -class ProcessMachine(BaseStateMachine): - """ - State machine for managing Process (OS subprocess) lifecycle. - - Process Lifecycle: - ┌─────────────────────────────────────────────────────────────┐ - │ QUEUED State │ - │ • Process ready to launch, waiting for resources │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() when can_start() - ┌─────────────────────────────────────────────────────────────┐ - │ RUNNING State → enter_running() │ - │ 1. process.launch() │ - │ • Spawn subprocess with cmd, pwd, env, timeout │ - │ • Set pid, started_at │ - │ • Process runs in background or foreground │ - │ 2. Monitor process completion │ - │ • Check exit code when process completes │ - └─────────────────────────────────────────────────────────────┘ - ↓ tick() checks is_exited() - ┌─────────────────────────────────────────────────────────────┐ - │ EXITED State │ - │ • Process completed (exit_code set) │ - │ • Health stats incremented │ - │ • stdout/stderr captured │ - └─────────────────────────────────────────────────────────────┘ - - Note: This is a simpler state machine than ArchiveResult. - Process is just about execution lifecycle. ArchiveResult handles - the archival-specific logic (status, output parsing, etc.). - """ - - model_attr_name = "process" - process: Process - - # States - queued = State(value=Process.StatusChoices.QUEUED, initial=True) - running = State(value=Process.StatusChoices.RUNNING) - exited = State(value=Process.StatusChoices.EXITED, final=True) - - # Tick Event - transitions based on conditions - tick = ( - queued.to.itself(unless="can_start") - | queued.to(running, cond="can_start") - | running.to.itself(unless="is_exited") - | running.to(exited, cond="is_exited") - ) - - # Additional events (for explicit control) - launch = queued.to(running) - kill = running.to(exited) - - def can_start(self) -> bool: - """Check if process can start (has cmd and machine).""" - return bool(self.process.cmd and self.process.machine) - - def is_exited(self) -> bool: - """Check if process has exited (exit_code is set).""" - return self.process.exit_code is not None - - @queued.enter - def enter_queued(self): - """Process is queued for execution.""" - self.process.update_and_requeue( - retry_at=timezone.now(), - status=Process.StatusChoices.QUEUED, - ) - - @running.enter - def enter_running(self): - """Start process execution.""" - # Lock the process while it runs - self.process.update_and_requeue( - retry_at=timezone.now() + timedelta(seconds=self.process.timeout), - status=Process.StatusChoices.RUNNING, - started_at=timezone.now(), - ) - - # Launch the subprocess - # NOTE: This is a placeholder - actual launch logic would - # be implemented based on how hooks currently spawn processes - # For now, Process is a data model that tracks execution metadata - # The actual subprocess spawning is still handled by run_hook() - - # Mark as immediately exited for now (until we refactor run_hook) - # In the future, this would actually spawn the subprocess - self.process.exit_code = 0 # Placeholder - self.process.save() - - @exited.enter - def enter_exited(self): - """Process has exited.""" - self.process.update_and_requeue( - retry_at=None, - status=Process.StatusChoices.EXITED, - ended_at=timezone.now(), - ) - - -# ============================================================================= -# State Machine Registration -# ============================================================================= - -# Manually register state machines with python-statemachine registry -registry.register(BinaryMachine) -registry.register(ProcessMachine) diff --git a/archivebox/misc/serve_static.py b/archivebox/misc/serve_static.py index c1b925df..3c27bce4 100644 --- a/archivebox/misc/serve_static.py +++ b/archivebox/misc/serve_static.py @@ -33,6 +33,26 @@ from archivebox.misc.logging_util import printable_filesize _HASHES_CACHE: dict[Path, tuple[float, dict[str, str]]] = {} IMG_SRC_ATTR_RE = re.compile(r'(]*?\s(?:src|data-src)=["\'])([^"\']+)(["\'])', re.IGNORECASE) TRANSFORMED_HTML_PREVIEW_STYLE = """