diff --git a/archivebox/api/urls.py b/archivebox/api/urls.py index 5ec44357..6a21bd75 100644 --- a/archivebox/api/urls.py +++ b/archivebox/api/urls.py @@ -8,7 +8,6 @@ from django.urls import path from django.views.generic.base import RedirectView from archivebox.core.host_utils import build_web_url - from .v1_api import urls as v1_api_urls diff --git a/archivebox/api/v1_crawls.py b/archivebox/api/v1_crawls.py index 54e695dc..771338d1 100644 --- a/archivebox/api/v1_crawls.py +++ b/archivebox/api/v1_crawls.py @@ -1,8 +1,9 @@ __package__ = "archivebox.api" +from pathlib import Path from uuid import UUID from datetime import datetime -from django.http import HttpRequest +from django.http import FileResponse, HttpRequest from django.shortcuts import redirect from django.utils import timezone @@ -15,7 +16,7 @@ from ninja.errors import HttpError from archivebox.core.models import Snapshot from archivebox.crawls.models import Crawl -from .auth import API_AUTH_METHODS +from .auth import API_AUTH_METHODS, auth_using_token router = Router(tags=["Crawl Models"], auth=API_AUTH_METHODS) @@ -117,6 +118,8 @@ def create_crawl(request: HttpRequest, data: CrawlCreateSchema): created_by=request.user if isinstance(request.user, User) else None, ) crawl.create_snapshots_from_urls() + if not crawl.snapshot_set.exists(): + crawl.sm.seal() return crawl @@ -136,6 +139,50 @@ def get_crawl(request: HttpRequest, crawl_id: str, as_rss: bool = False, with_sn return crawl +def crawl_file(request: HttpRequest, crawl_id: str, path: str): + user = getattr(request, "user", None) + is_superuser = bool( + getattr(user, "is_authenticated", False) and getattr(user, "is_active", False) and getattr(user, "is_superuser", False), + ) + if not is_superuser: + token = request.GET.get("api_key") or request.headers.get("X-ArchiveBox-API-Key") + auth_header = request.headers.get("Authorization", "") + if not token and auth_header.lower().startswith("bearer "): + token = auth_header.split(None, 1)[1].strip() + token_user = auth_using_token(token=token, request=request) if token else None + is_superuser = bool(token_user and token_user.is_active and token_user.is_superuser) + if not is_superuser: + raise HttpError(403, "Permission denied") + + crawl = Crawl.objects.get(id__icontains=crawl_id) + crawl_root = Path(crawl.output_dir).resolve() + file_path = (crawl_root / path).resolve() + if not file_path.is_file() or crawl_root not in file_path.parents: + raise HttpError(404, "Crawl file not found") + + response = FileResponse(file_path.open("rb")) + response["Cache-Control"] = "no-store, no-cache, max-age=0, must-revalidate" + response["Pragma"] = "no-cache" + response["Expires"] = "0" + response["X-Content-Type-Options"] = "nosniff" + return response + + +@router.get("/crawl/{crawl_id}/files/{filename}", auth=None, url_name="crawl_file_root") +def crawl_file_root(request: HttpRequest, crawl_id: str, filename: str): + return crawl_file(request, crawl_id, filename) + + +@router.get("/crawl/{crawl_id}/files/{folder}/{filename}", auth=None, url_name="crawl_file_nested_1") +def crawl_file_nested_1(request: HttpRequest, crawl_id: str, folder: str, filename: str): + return crawl_file(request, crawl_id, f"{folder}/{filename}") + + +@router.get("/crawl/{crawl_id}/files/{folder}/{subfolder}/{filename}", auth=None, url_name="crawl_file_nested_2") +def crawl_file_nested_2(request: HttpRequest, crawl_id: str, folder: str, subfolder: str, filename: str): + return crawl_file(request, crawl_id, f"{folder}/{subfolder}/{filename}") + + @router.patch("/crawl/{crawl_id}", response=CrawlSchema, url_name="patch_crawl") def patch_crawl(request: HttpRequest, crawl_id: str, data: CrawlUpdateSchema): """Update a crawl (e.g., set status=sealed to cancel queued work).""" diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py index 950f3061..255f15d5 100644 --- a/archivebox/cli/archivebox_add.py +++ b/archivebox/cli/archivebox_add.py @@ -197,10 +197,14 @@ def add( # Just create the crawl but don't start processing print("[yellow]\\[*] Index-only mode - crawl created but not started[/yellow]") crawl.create_snapshots_from_urls() + if not crawl.snapshot_set.exists(): + crawl.sm.seal() return crawl, crawl.snapshot_set.all() if bg: crawl.create_snapshots_from_urls() + if not crawl.snapshot_set.exists(): + crawl.sm.seal() # 5. Start the crawl runner to process the queue # The runner will: @@ -225,10 +229,14 @@ def add( exit_code = 0 try: try: - with foreground_shutdown_signals(), foreground_parent_watchdog(): + with foreground_shutdown_signals(first_signal_message=None), foreground_parent_watchdog(): while True: standby_until_runtime_stack_needed(command, data_dir=CONSTANTS.DATA_DIR) - exit_code = run_runner_worker(["--crawl-id", str(crawl.id)], name=f"worker_runner_add_{os.getpid()}") + exit_code = run_runner_worker( + ["--crawl-id", str(crawl.id)], + name=f"worker_runner_add_{os.getpid()}", + interactive_interrupts=True, + ) if exit_code == 0: break if not command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR): diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 7ff05752..8730b426 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -263,9 +263,15 @@ def run_runner(daemon: bool = False, crawl_id: str | None = None, maintenance_on rprint(f"[green][*] Existing ArchiveBox orchestrator pid={existing_pid} is already running.[/green]", file=sys.stderr) return 0 current.mark_running(process_type=Process.TypeChoices.ORCHESTRATOR, pwd=str(CONSTANTS.DATA_DIR), timeout=0) + interactive_interrupts = current.root.process_type == Process.TypeChoices.ADD try: with foreground_shutdown_signals(), foreground_parent_watchdog(enabled=not daemon): - run_pending_crawls(daemon=daemon, crawl_id=crawl_id, maintenance_only=maintenance_only) + run_pending_crawls( + daemon=daemon, + crawl_id=crawl_id, + maintenance_only=maintenance_only, + interactive_interrupts=interactive_interrupts, + ) return 0 except (KeyboardInterrupt, asyncio.CancelledError): return 0 diff --git a/archivebox/cli/archivebox_server.py b/archivebox/cli/archivebox_server.py index d8a351c8..0d39d403 100644 --- a/archivebox/cli/archivebox_server.py +++ b/archivebox/cli/archivebox_server.py @@ -4,6 +4,9 @@ __package__ = "archivebox.cli" import sys import os +import socket +import subprocess +import time from collections.abc import Iterable import rich_click as click @@ -22,7 +25,7 @@ def server( nothreading: bool = False, ) -> None: """Run the ArchiveBox HTTP server""" - from archivebox.config.common import get_config, rprint + from archivebox.config.common import get_config config = get_config() runserver_args = list(runserver_args or (config.BIND_ADDR,)) @@ -62,12 +65,53 @@ def server( except IndexError: pass + if daemonize and os.environ.get("ARCHIVEBOX_SERVER_DAEMON_CHILD") != "1": + from archivebox.config import CONSTANTS + + log_path = CONSTANTS.LOGS_DIR / "server.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + daemon_env = os.environ.copy() + daemon_env["ARCHIVEBOX_SERVER_DAEMON_CHILD"] = "1" + daemon_cmd = [sys.executable, "-m", "archivebox", "server"] + if debug: + daemon_cmd.append("--debug") + if reload: + daemon_cmd.append("--reload") + if nothreading: + daemon_cmd.append("--nothreading") + daemon_cmd.extend(runserver_args) + with log_path.open("a", encoding="utf-8") as log_file: + proc = subprocess.Popen( + daemon_cmd, + cwd=os.getcwd(), + env=daemon_env, + stdin=subprocess.DEVNULL, + stdout=log_file, + stderr=log_file, + start_new_session=True, + ) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if proc.poll() is not None: + print(f"[red][X] ArchiveBox daemon server exited early with code {proc.returncode}. See {log_path}[/red]") + sys.exit(proc.returncode or 1) + try: + with socket.create_connection((host, int(port)), timeout=0.25): + break + except OSError: + time.sleep(0.1) + else: + print(f"[yellow][!] ArchiveBox daemon server pid={proc.pid} is still starting. See {log_path}[/yellow]") + return + os.environ["BIND_ADDR"] = f"{host}:{port}" from archivebox.core.host_utils import build_admin_url admin_url = build_admin_url("/admin/") from archivebox.workers.supervisord_util import ( + active_supervisord_runtime_components, + format_runtime_components, start_server_workers, stop_existing_supervisord_process, is_port_in_use, @@ -105,11 +149,16 @@ def server( connections.close_all() try: - with foreground_shutdown_signals(), foreground_parent_watchdog(): + with foreground_shutdown_signals(), foreground_parent_watchdog(enabled=os.environ.get("ARCHIVEBOX_SERVER_DAEMON_CHILD") != "1"): while True: - standby_until_runtime_stack_needed(command, data_dir=config.DATA_DIR) - sys.stdout.write(f"[*] ArchiveBox server parent pid={os.getpid()} is now running the orchestrator and server...\n") - sys.stdout.flush() + standby_result = standby_until_runtime_stack_needed(command, data_dir=config.DATA_DIR) + older_owner = runtime_stack_owner(data_dir=config.DATA_DIR, exclude_id=command.id) + takeover_components = active_supervisord_runtime_components(config=config) + if older_owner and takeover_components: + print( + "[yellow][*] Taking over " + f"{format_runtime_components(takeover_components)} from older existing archivebox process (pid={older_owner.pid}).[/yellow]", + ) stop_existing_supervisord_process() if is_port_in_use(host, int(port)): print(f"[red][X] Error: Port {port} is already in use[/red]") @@ -119,23 +168,17 @@ def server( result = start_server_workers( host=host, port=port, - daemonize=daemonize, + daemonize=False, debug=run_in_debug, reload=reload, nothreading=nothreading, keep_running=still_owns_runtime_stack, should_stop_supervisord=still_owns_runtime_stack, + resumed_from_pid=standby_result.get("previous_owner_pid") if standby_result.get("resumed") else None, ) if result == "interrupted": break if not still_owns_runtime_stack(): - owner = runtime_stack_owner(data_dir=config.DATA_DIR) - owner_pid = owner.pid if owner else "unknown" - rprint( - "[yellow][*] A newer archivebox process took over the runner " - f"(pid={owner_pid}). Work will continue there, and will continue here if the other process is stopped and work still remains.[/yellow]", - file=sys.stderr, - ) continue if result == "exited": print("[yellow][*] Runtime stack exited while this parent is still leader; restarting...[/yellow]") diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index 9198d5dd..35618509 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -533,11 +533,9 @@ def drain_old_archive_dirs(resume_from: str | None = None, batch_size: int = 500 try: snapshot.status = Snapshot.StatusChoices.SEALED snapshot.retry_at = timezone.now() - Snapshot.objects.bulk_create([snapshot]) - Snapshot.objects.filter(pk=snapshot.pk).update( - status=Snapshot.StatusChoices.SEALED, - retry_at=snapshot.retry_at, - ) + # Snapshot.save() owns URL validation and filesystem/index side + # effects. Do not use bulk_create() here; it bypasses save(). + snapshot.save() crawl = _get_snapshot_crawl(snapshot) if crawl is not None: diff --git a/archivebox/core/migrations/0045_archiveresult_unique_hook.py b/archivebox/core/migrations/0045_archiveresult_unique_hook.py new file mode 100644 index 00000000..4c34c808 --- /dev/null +++ b/archivebox/core/migrations/0045_archiveresult_unique_hook.py @@ -0,0 +1,17 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0044_alter_archiveresult_status_alter_snapshot_status"), + ] + + operations = [ + migrations.AddConstraint( + model_name="archiveresult", + constraint=models.UniqueConstraint( + fields=("snapshot", "plugin", "hook_name"), + name="unique_archiveresult_per_snapshot_hook", + ), + ), + ] diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 76b640d6..667cb999 100755 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -1342,8 +1342,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW self.tags.add(tag) def _merge_archive_results_from_index(self, index_data: dict, update_existing: bool = True): - """Merge ArchiveResults - keep both (by plugin+start_ts).""" - existing = {(ar.plugin, ar.start_ts): ar for ar in ArchiveResult.objects.filter(snapshot=self)} + """Merge ArchiveResults one row per hook; retries update the existing row.""" + existing = {(ar.plugin, ar.hook_name): ar for ar in ArchiveResult.objects.filter(snapshot=self)} if update_existing: for archiveresult in existing.values(): normalized_status = ArchiveResult.normalize_status(archiveresult.status) @@ -1404,7 +1404,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW output_json = result_data.get("output_json") output_mimetypes = result_data.get("output_mimetypes", "") - existing_result = existing.get((plugin, start_ts)) + hook_name = result_data.get("hook_name", "") + existing_result = existing.get((plugin, hook_name)) if existing_result: if not update_existing: return @@ -1413,22 +1414,25 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW if existing_result.status != status: existing_result.status = status update_fields.append("status") - if output_str and not existing_result.output_str: + if output_str and existing_result.output_str != output_str: existing_result.output_str = output_str update_fields.append("output_str") - if output_json and not existing_result.output_json: + if output_json and existing_result.output_json != output_json: existing_result.output_json = output_json update_fields.append("output_json") - if output_files and not existing_result.output_files: + if output_files and existing_result.output_files != output_files: existing_result.output_files = output_files update_fields.append("output_files") - if output_size and not existing_result.output_size: + if "output_size" in result_data and existing_result.output_size != output_size: existing_result.output_size = output_size update_fields.append("output_size") - if output_mimetypes and not existing_result.output_mimetypes: + if output_mimetypes and existing_result.output_mimetypes != output_mimetypes: existing_result.output_mimetypes = output_mimetypes update_fields.append("output_mimetypes") - if end_ts and not existing_result.end_ts: + if start_ts and existing_result.start_ts != start_ts: + existing_result.start_ts = start_ts + update_fields.append("start_ts") + if end_ts and existing_result.end_ts != end_ts: existing_result.end_ts = end_ts update_fields.append("end_ts") if update_fields: @@ -1455,7 +1459,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW archiveresult = ArchiveResult.objects.create( snapshot=self, plugin=plugin, - hook_name=result_data.get("hook_name", ""), + hook_name=hook_name, status=status, output_str=output_str, output_json=output_json, @@ -1466,7 +1470,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW end_ts=end_ts, process=process, ) - existing[(plugin, start_ts)] = archiveresult + existing[(plugin, hook_name)] = archiveresult def write_index_json(self): """Write index.json in 0.9.x format (deprecated, use write_index_jsonl).""" @@ -2447,15 +2451,13 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW hook_name = hook_path.name # e.g., 'on_Snapshot__50_wget.py' plugin = hook_path.parent.name # e.g., 'wget' - # Check if AR already exists for this specific hook - if ArchiveResult.objects.filter(snapshot=self, hook_name=hook_name).exists(): - continue - - archiveresult, created = ArchiveResult.objects.get_or_create( + # ArchiveResult output is one filesystem directory per plugin hook, so + # retries must update this row in place instead of creating siblings. + archiveresult, _created = ArchiveResult.objects.get_or_create( snapshot=self, + plugin=plugin, hook_name=hook_name, defaults={ - "plugin": plugin, "status": ArchiveResult.INITIAL_STATE, }, ) @@ -3277,6 +3279,9 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, M models.Index(fields=["snapshot", "status"], name="archiveresult_snap_status_idx"), models.Index(fields=["-start_ts", "-id"], name="archiveresult_start_idx"), ] + constraints = [ + models.UniqueConstraint(fields=["snapshot", "plugin", "hook_name"], name="unique_archiveresult_per_snapshot_hook"), + ] def __str__(self): return f"[{self.id}] {self.snapshot.url[:64]} -> {self.plugin}" @@ -3381,15 +3386,16 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, M except ArchiveResult.DoesNotExist: pass - # Get or create by snapshot_id + plugin + # Get or create by snapshot_id + plugin + hook_name. The filesystem has a + # single output dir for each hook, so retries update that same DB row. try: snapshot = Snapshot.objects.get(id=snapshot_id) result, _ = ArchiveResult.objects.get_or_create( snapshot=snapshot, plugin=plugin, + hook_name=record.get("hook_name", ""), defaults={ - "hook_name": record.get("hook_name", ""), "status": record.get("status", "queued"), "output_str": record.get("output_str", ""), }, diff --git a/archivebox/core/shutdown_util.py b/archivebox/core/shutdown_util.py index 3e621131..748ee13e 100644 --- a/archivebox/core/shutdown_util.py +++ b/archivebox/core/shutdown_util.py @@ -98,6 +98,8 @@ def kill_remaining_processes(processes: list[psutil.Process], *, timeout: float @contextmanager def foreground_shutdown_signals( handled_signals: tuple[signal.Signals, ...] = (signal.SIGHUP, signal.SIGINT, signal.SIGTERM), + *, + first_signal_message: str | None = "\n[πŸ›‘] Got {signal_name}, stopping gracefully...\n", ) -> Iterator[ShutdownSignalState]: """Install foreground signal handlers that print an immediate exit notice. @@ -115,7 +117,8 @@ def foreground_shutdown_signals( def raise_keyboard_interrupt(signum, _frame): if state.signal_name is None: state.signal_name = signal.Signals(signum).name - os.write(sys.stdout.fileno(), f"\n[πŸ›‘] Got {state.signal_name}, stopping gracefully...\n".encode()) + if first_signal_message is not None: + os.write(sys.stdout.fileno(), first_signal_message.format(signal_name=state.signal_name).encode()) raise KeyboardInterrupt try: diff --git a/archivebox/core/urls.py b/archivebox/core/urls.py index 23117cf1..819789f0 100644 --- a/archivebox/core/urls.py +++ b/archivebox/core/urls.py @@ -24,7 +24,6 @@ from archivebox.core.views import ( AddView, WebAddView, HealthCheckView, - live_progress_screencast_frame_view, live_progress_view, ) @@ -70,11 +69,6 @@ urlpatterns = [ path("accounts/login/", RedirectView.as_view(url="/admin/login/")), path("accounts/logout/", RedirectView.as_view(url="/admin/logout/")), path("accounts/", include("django.contrib.auth.urls")), - re_path( - r"^admin/live-progress/screencast/(?P[0-9a-fA-F-]{8,36})\.jpg$", - archivebox_admin.admin_view(live_progress_screencast_frame_view), - name="live_progress_screencast_frame", - ), path("admin/live-progress/", archivebox_admin.admin_view(live_progress_view), name="live_progress"), path("admin/", archivebox_admin.urls), path("api/", include("archivebox.api.urls"), name="api"), diff --git a/archivebox/core/views.py b/archivebox/core/views.py index c35798df..0b8e717a 100644 --- a/archivebox/core/views.py +++ b/archivebox/core/views.py @@ -12,7 +12,7 @@ from pathlib import Path from urllib.parse import quote, urlparse from django.shortcuts import render, redirect -from django.http import FileResponse, JsonResponse, HttpRequest, HttpResponse, Http404, HttpResponseForbidden, QueryDict +from django.http import JsonResponse, HttpRequest, HttpResponse, Http404, HttpResponseForbidden, QueryDict from django.utils.html import format_html from django.utils.safestring import mark_safe from django.views import View @@ -22,7 +22,6 @@ from django.db.models import CharField, Count, Q, Prefetch, Sum from django.db.models.functions import Cast from django.contrib import messages from django.contrib.auth.mixins import UserPassesTestMixin -from django.core.signing import BadSignature, SignatureExpired, TimestampSigner from django.views.decorators.csrf import csrf_exempt from django.views.decorators.gzip import gzip_page from django.utils.decorators import method_decorator @@ -79,7 +78,6 @@ from archivebox.hooks import ( ABX_PLUGINS_GITHUB_BASE_URL = "https://github.com/ArchiveBox/abx-plugins/tree/main/abx_plugins/plugins/" LIVE_PLUGIN_BASE_URL = "/admin/environment/plugins/" -SCREENCAST_SIGNER = TimestampSigner(salt="archivebox.live-progress.screencast") def _get_request_config(request: HttpRequest, *, resolve_plugins: bool = False): @@ -1257,9 +1255,12 @@ class AddView(UserPassesTestMixin, FormView): crawl.save(update_fields=["schedule"]) crawl.create_snapshots_from_urls() - from archivebox.services.runner import ensure_background_runner + if crawl.snapshot_set.exists(): + from archivebox.services.runner import ensure_background_runner - ensure_background_runner() + ensure_background_runner() + else: + crawl.sm.seal() return crawl @@ -1386,33 +1387,6 @@ class HealthCheckView(View): return HttpResponse("OK", content_type="text/plain", status=200) -def live_progress_screencast_frame_view(request, object_id: str): - """Serve cache-only Chrome screencast frames through the admin app.""" - if not is_admin_user(request): - return HttpResponseForbidden("Permission denied") - - token = request.GET.get("token", "") - try: - if SCREENCAST_SIGNER.unsign(token, max_age=60) != str(object_id): - return HttpResponseForbidden("Permission denied") - except (BadSignature, SignatureExpired): - return HttpResponseForbidden("Permission denied") - - live_root = (CONSTANTS.CACHE_DIR / "chrome_screencast").resolve() - frame_path = live_root / str(object_id) / "latest.jpg" - try: - resolved_frame_path = frame_path.resolve(strict=True) - except FileNotFoundError: - raise Http404 from None - if not resolved_frame_path.is_file() or live_root not in resolved_frame_path.parents: - raise Http404 - - response = FileResponse(resolved_frame_path.open("rb"), content_type="image/jpeg") - response["Cache-Control"] = "no-store, max-age=0" - response["X-Content-Type-Options"] = "nosniff" - return response - - @gzip_page def live_progress_view(request): """Simple JSON endpoint for live progress status - used by admin progress monitor.""" @@ -1560,16 +1534,17 @@ def live_progress_view(request): url = str(url or "") return url if len(url) <= 96 else f"{url[:93]}..." - def screencast_frame_url(object_id: str) -> str: - frame_path = CONSTANTS.CACHE_DIR / "chrome_screencast" / object_id / "latest.jpg" + def screencast_frame_url(crawl_id: str, crawl_dir: Path) -> str: + frame_path = crawl_dir / "chrome_screencast" / "latest.jpg" try: frame_stat = frame_path.stat() except OSError: return "" if frame_stat.st_size <= 0: return "" - token = SCREENCAST_SIGNER.sign(object_id) - return f"/admin/live-progress/screencast/{object_id}.jpg?v={frame_stat.st_mtime_ns}&token={quote(token)}" + if now.timestamp() - frame_stat.st_mtime > 15: + return "" + return f"/api/v1/crawls/crawl/{crawl_id}/files/chrome_screencast/latest.jpg" machine_id = Machine.current().id orchestrator_proc = ( @@ -1982,7 +1957,7 @@ def live_progress_view(request): crawl_setup_completed = sum(1 for item in crawl_setup_plugins if item.get("status") == "succeeded") crawl_setup_failed = sum(1 for item in crawl_setup_plugins if item.get("status") == "failed") crawl_setup_pending = sum(1 for item in crawl_setup_plugins if item.get("status") == "queued") - crawl_screencast_url = screencast_frame_url(crawl_id) + crawl_screencast_url = screencast_frame_url(crawl_id, active_crawl_objects[crawl_id].output_dir) crawl_screencast_link = f"/admin/crawls/crawl/{crawl_id}/change/" if crawl_screencast_url else "" # Get active snapshots for this crawl (already prefetched) @@ -2034,7 +2009,7 @@ def live_progress_view(request): snapshot_preview_url = snapshot_favicon_url if snapshot["status"] == Snapshot.StatusChoices.STARTED: - snapshot_screencast_url = screencast_frame_url(str(snapshot["id"])) + snapshot_screencast_url = screencast_frame_url(crawl_id, active_crawl_objects[crawl_id].output_dir) snapshot_screencast_link = snapshot_view_url(snapshot) if snapshot_screencast_url else "" def plugin_sort_key(ar): diff --git a/archivebox/crawls/models.py b/archivebox/crawls/models.py index 713c2981..a95752a6 100755 --- a/archivebox/crawls/models.py +++ b/archivebox/crawls/models.py @@ -887,7 +887,7 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith snapshot.save(update_fields=["depth", "title", "timestamp", "status", "retry_at", "url", "crawl", "modified_at"]) else: snapshot = Snapshot(id=snapshot_id, url=url, crawl=self, **defaults) - snapshot.save(force_insert=True) + snapshot.save() created = True else: snapshot = Snapshot.objects.filter(url=url, crawl=self).first() @@ -896,7 +896,7 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith else: try: snapshot = Snapshot(url=url, crawl=self, **defaults) - snapshot.save(force_insert=True) + snapshot.save() created = True except IntegrityError: snapshot = Snapshot.objects.get(url=url, crawl=self) @@ -1052,10 +1052,19 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith for snapshot in snapshots: snapshot.set_delete_at_from_config(config.DELETE_AFTER) - try: - created_snapshots = list(Snapshot.objects.bulk_create(snapshots)) - except ValidationError as err: - print(f"[yellow][!] Skipping blocked discovered snapshots: {err}[/yellow]") + created_snapshots = [] + for snapshot in snapshots: + try: + # Snapshot.save() owns URL validation and filesystem/index side + # effects. Do not use bulk_create() here; it bypasses save(). + snapshot.save() + except IntegrityError: + continue + except ValidationError as err: + print(f"[yellow][!] Skipping blocked discovered snapshot URL: {snapshot.url} ({err})[/yellow]") + continue + created_snapshots.append(snapshot) + if not created_snapshots: return [] crawl_urls = {url for _raw_line, url in self._iter_url_lines() if url} @@ -1072,8 +1081,9 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith } if tag_names: tag_names_by_url[snapshot.url] = tag_names - # Same transaction rule as create_snapshots_from_urls(): bulk_create() - # only writes DB rows; symlink creation waits until after commit. + # Snapshot.save() handles model-level validation. The crawl symlink + # can still wait until after commit so SQLite does not hold a write + # lock while touching the filesystem. transaction.on_commit(lambda snapshot=snapshot: snapshot.ensure_crawl_symlink()) tag_names = {tag for tags in tag_names_by_url.values() for tag in tags} @@ -1469,8 +1479,10 @@ class CrawlMachine(BaseStateMachine): | paused.to.itself() ) - # Manual event (triggered by last Snapshot sealing) - seal = started.to(sealed) + # 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) pause_requested = queued.to(paused) | started.to(paused) resume_requested = paused.to(queued) diff --git a/archivebox/services/archive_result_service.py b/archivebox/services/archive_result_service.py index c53b46cb..5917441b 100644 --- a/archivebox/services/archive_result_service.py +++ b/archivebox/services/archive_result_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import json from collections import defaultdict from collections.abc import Iterable @@ -209,6 +210,8 @@ class ArchiveResultService(BaseService): EMITS = [] def __init__(self, bus): + self._completed_process_event_ids: set[str] = set() + self._save_locks: dict[tuple[str, str, str], asyncio.Lock] = {} super().__init__(bus) self.bus.on(ArchiveResultEvent, self.on_ArchiveResultEvent__save_to_db) self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db) @@ -259,12 +262,15 @@ class ArchiveResultService(BaseService): if event.error: defaults["notes"] = event.error - result, _created = await ArchiveResult.objects.aupdate_or_create( - snapshot=snapshot, - plugin=event.plugin, - hook_name=event.hook_name, - defaults=defaults, - ) + key = (str(snapshot.id), event.plugin, event.hook_name) + lock = self._save_locks.setdefault(key, asyncio.Lock()) + async with lock: + result, _created = await ArchiveResult.objects.aupdate_or_create( + snapshot=snapshot, + plugin=event.plugin, + hook_name=event.hook_name, + defaults=defaults, + ) if result.status in (ArchiveResult.StatusChoices.SUCCEEDED, ArchiveResult.StatusChoices.NORESULTS): next_title = _extract_snapshot_title(str(snapshot.output_dir), event.plugin, result.output_str, snapshot_url=snapshot.url) @@ -273,6 +279,10 @@ class ArchiveResultService(BaseService): await snapshot.asave(update_fields=["title", "modified_at"]) async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None: + if event.event_id in self._completed_process_event_ids: + return + self._completed_process_event_ids.add(event.event_id) + if not event.hook_name.startswith("on_Snapshot"): return snapshot_event = await self.bus.find( @@ -286,6 +296,10 @@ class ArchiveResultService(BaseService): records = _iter_archiveresult_records(event.stdout) if records: + if len(records) > 1: + raise RuntimeError( + f"Hook {event.plugin_name}:{event.hook_name} emitted {len(records)} ArchiveResult records; expected exactly one", + ) for record in records: record_status = _normalize_status(record.get("status") or "") record_failed = record_status == "failed" or (not record_status and event.exit_code not in (0, PROCESS_EXIT_SKIPPED)) diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index d9cd382b..df655d51 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -20,6 +20,7 @@ from typing import Any from asgiref.sync import sync_to_async from django.utils import timezone from rich.console import Console +from rich.text import Text from abx_dl.events import ( BinaryRequestEvent, @@ -75,6 +76,43 @@ def _bus_name(prefix: str, identifier: str) -> str: return f"{prefix}_{normalized}" +def _runner_short_id(identifier) -> str: + return str(identifier).replace("-", "")[-8:] + + +def _runner_label(value: str, *, reserve: int) -> str: + width = max(24, shutil.get_terminal_size(fallback=(120, 40)).columns - reserve) + value = " ".join(str(value or "").split()) + if len(value) <= width: + return value + return f"{value[: max(0, width - 3)]}..." + + +def _runner_console_line(*, crawl, snapshot=None, status: str = "STARTED") -> None: + line = Text() + line.append(f"[Crawl#{_runner_short_id(crawl.id)}]", style="cyan bold") + line.append(" ") + if snapshot is not None: + line.append(f"[Snapshot#{_runner_short_id(snapshot.id)}]", style="magenta bold") + line.append(" ") + status_styles = { + "STARTED": "green bold", + "SEALED": "blue bold", + "PAUSED": "yellow bold", + } + line.append(f"[{status}]", style=status_styles.get(status, "white bold")) + line.append(" ") + prefix_width = len(line.plain) + if snapshot is not None: + label = snapshot.url + else: + label = (getattr(crawl, "label", "") or "").strip() + if not label: + label = (getattr(crawl, "urls", "") or "").partition("\n")[0].strip() or str(crawl.id) + line.append(_runner_label(label, reserve=prefix_width)) + Console(highlight=False).print(line) + + def _count_selected_hooks(plugins: dict[str, Plugin], selected_plugins: list[str] | None) -> int: selected = filter_plugins(plugins, selected_plugins) if selected_plugins else plugins return sum(1 for plugin in selected.values() for hook in plugin.hooks if "CrawlSetup" in hook.name or "Snapshot" in hook.name) @@ -137,6 +175,12 @@ async def _run_event_now(event, timeout: float | None = None): return event +async def _emit_event_now(bus, event, timeout: float | None = None, parent_event=None): + if parent_event is not None: + event.event_parent_id = parent_event.event_id + return await _run_event_now(bus.emit(event), timeout) + + def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool: if os.environ.get("PYTEST_CURRENT_TEST") and not allow_under_pytest: return False @@ -186,11 +230,12 @@ class CrawlRunner: selected_plugins: list[str] | None = None, process_discovered_snapshots_inline: bool = True, show_progress: bool = True, + interactive_interrupts: bool = False, ): self.crawl = crawl self.bus = create_bus(name=_bus_name("ArchiveBox", str(crawl.id)), total_timeout=3600.0) self.plugins = discover_plugins() - HookProcessService(self.bus, emit_jsonl=False, interactive_tty=False) + HookProcessService(self.bus, emit_jsonl=False, interactive_tty=interactive_interrupts) register_sonic_daemon_event_handler(self.bus) PersistedProcessService(self.bus) BinaryService(self.bus) @@ -199,6 +244,7 @@ class CrawlRunner: MachineService(self.bus) self.process_discovered_snapshots_inline = process_discovered_snapshots_inline self.show_progress = show_progress + self.interactive_interrupts = interactive_interrupts async def ignore_snapshot(_snapshot_id: str) -> None: return None @@ -236,8 +282,17 @@ class CrawlRunner: previous = signal.getsignal(sig) def request_abort(_signum, _frame, sig=sig) -> None: - os.write(sys.stdout.fileno(), f"\n[πŸ›‘] Got {sig.name}, stopping gracefully...\n".encode()) + already_requested = self._signal_abort_requested + if not already_requested: + message = ( + f"\n[πŸ›‘] Got {sig.name}, aborting the active hook...\n" + if self.interactive_interrupts + else f"\n[πŸ›‘] Got {sig.name}, stopping gracefully...\n" + ) + os.write(sys.stdout.fileno(), message.encode()) self._request_abort_from_signal(sig) + if not already_requested: + return raise KeyboardInterrupt signal.signal(sig, request_abort) @@ -251,9 +306,10 @@ class CrawlRunner: signal.signal(sig, previous) def _request_abort_from_signal(self, _sig: signal.Signals) -> None: + already_requested = self._signal_abort_requested self._signal_abort_requested = True self._skip_wait_until_idle = True - if self._run_task is not None and not self._run_task.done(): + if (not self.interactive_interrupts or already_requested) and self._run_task is not None and not self._run_task.done(): self._run_task.cancel() async def crawl_is_cancelled(self) -> bool: @@ -846,30 +902,40 @@ class CrawlRunner: finally: if self.snapshot_tasks: await self.drain_snapshot_tasks() - await _run_event_now( - event.emit( - CrawlCleanupEvent( - url=snapshot["url"], - snapshot_id=snapshot["id"], - output_dir=str(output_dir), - event_timeout=crawl_setup_phase_timeout, - event_handler_slow_timeout=slow_warning_timeout(crawl_setup_phase_timeout), + await asyncio.shield( + asyncio.create_task( + _emit_event_now( + self.bus, + CrawlCleanupEvent( + url=snapshot["url"], + snapshot_id=snapshot["id"], + output_dir=str(output_dir), + event_timeout=crawl_setup_phase_timeout, + event_handler_slow_timeout=slow_warning_timeout(crawl_setup_phase_timeout), + ), + crawl_setup_phase_timeout, + parent_event=event, ), + context=contextvars.Context(), ), - crawl_setup_phase_timeout, ) finally: cancel_watcher.cancel() await asyncio.gather(cancel_watcher, return_exceptions=True) - await _run_event_now( - event.emit( - CrawlCompletedEvent( - url=snapshot["url"], - snapshot_id=snapshot["id"], - output_dir=str(output_dir), + await asyncio.shield( + asyncio.create_task( + _emit_event_now( + self.bus, + CrawlCompletedEvent( + url=snapshot["url"], + snapshot_id=snapshot["id"], + output_dir=str(output_dir), + ), + CrawlCompletedEvent.model_fields["event_timeout"].default, + parent_event=event, ), + context=contextvars.Context(), ), - CrawlCompletedEvent.model_fields["event_timeout"].default, ) on_archivebox_CrawlStartEvent.__name__ = "on_archivebox_CrawlStartEvent__run_snapshots" @@ -1046,6 +1112,7 @@ def run_crawl( selected_plugins: list[str] | None = None, process_discovered_snapshots_inline: bool = True, show_progress: bool = True, + interactive_interrupts: bool = False, ) -> None: from archivebox.crawls.models import Crawl from django.db import close_old_connections @@ -1061,6 +1128,7 @@ def run_crawl( selected_plugins=selected_plugins, process_discovered_snapshots_inline=process_discovered_snapshots_inline, show_progress=show_progress, + interactive_interrupts=interactive_interrupts, ).run(), ) finally: @@ -1197,9 +1265,9 @@ def run_snapshot_maintenance(snapshot_id: str) -> bool: return True -def run_due_crawl(crawl, *, lock_seconds: int) -> bool: +def run_due_crawl(crawl, *, lock_seconds: int, interactive_interrupts: bool = False) -> bool: if crawl.is_paused: - print(f"[runner] Crawl {str(crawl.id)[-12:]} paused; skipping until resumed", flush=True) + _runner_console_line(crawl=crawl, status="PAUSED") return True if crawl.status in (crawl.StatusChoices.QUEUED, crawl.StatusChoices.STARTED): from archivebox.core.models import Snapshot @@ -1255,29 +1323,28 @@ def run_due_crawl(crawl, *, lock_seconds: int) -> bool: return True if not crawl.claim_processing_lock(lock_seconds=lock_seconds): return False - print(f"[runner] Crawl {str(crawl.id)[-12:]} running status={crawl.status} snapshots={snapshot_count}", flush=True) - run_crawl(str(crawl.id), process_discovered_snapshots_inline=True) + _runner_console_line(crawl=crawl) + run_crawl(str(crawl.id), process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts) return True if crawl.status == crawl.StatusChoices.SEALED: - print(f"[runner] Crawl {str(crawl.id)[-12:]} sealed; clearing retry tick", flush=True) + _runner_console_line(crawl=crawl, status="SEALED") crawl.retry_at = None crawl.save(update_fields=["retry_at", "modified_at"]) return True - print(f"[runner] Crawl {str(crawl.id)[-12:]} status={crawl.status}; clearing retry tick", flush=True) crawl.retry_at = None crawl.save(update_fields=["retry_at", "modified_at"]) return True -def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool: +def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: bool = False) -> bool: from archivebox.core.models import Snapshot if snapshot.is_paused: selected_plugins = queued_plugins_for_snapshot(str(snapshot.id)) if snapshot.fs_migration_needed and Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds): - print(f"[runner] Snapshot {str(snapshot.id)[-12:]} paused maintenance fs_version={snapshot.fs_version}", flush=True) + _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot) run_snapshot_maintenance(str(snapshot.id)) if not selected_plugins: # No targeted plugin rows remain, so put paused snapshots back @@ -1298,10 +1365,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool: if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds): return False try: - print( - f"[runner] Snapshot {str(snapshot.id)[-12:]} paused targeted plugins={','.join(selected_plugins)} url={snapshot.url}", - flush=True, - ) + _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot) # Explicit maintenance, e.g. `archivebox update --index-only`, may # need to run search/index hooks for a paused snapshot. That should # not resume the crawl or make unrelated queued work runnable, so @@ -1312,6 +1376,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool: snapshot_ids=[str(snapshot.id)], selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, + interactive_interrupts=interactive_interrupts, ) finally: # Targeted plugin rows can complete while the Snapshot remains @@ -1328,22 +1393,17 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool: # migration. Run the filesystem/json save path before queued search # backfill rows so both maintenance streams stay ordered without # changing Snapshot.status away from SEALED. - print( - f"[runner] Snapshot {str(snapshot.id)[-12:]} sealed maintenance fs_version={snapshot.fs_version} url={snapshot.url}", - flush=True, - ) + _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot) return run_snapshot_maintenance(str(snapshot.id)) selected_plugins = queued_plugins_for_snapshot(str(snapshot.id)) if selected_plugins: - print( - f"[runner] Snapshot {str(snapshot.id)[-12:]} sealed targeted plugins={','.join(selected_plugins)} url={snapshot.url}", - flush=True, - ) + _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot) run_crawl( str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, + interactive_interrupts=interactive_interrupts, ) return True return run_snapshot_maintenance(str(snapshot.id)) @@ -1351,18 +1411,18 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool: if snapshot.status == Snapshot.StatusChoices.STARTED: _reset_count, running_count = reset_abandoned_snapshot_results(snapshot) if running_count: - print(f"[runner] Snapshot {str(snapshot.id)[-12:]} still has {running_count} running ArchiveResults", flush=True) snapshot.update_and_requeue(retry_at=timezone.now() + timedelta(seconds=ACTIVE_STATE_LEASE_SECONDS)) return True if not snapshot.claim_processing_lock(lock_seconds=lock_seconds): return False - print(f"[runner] Snapshot {str(snapshot.id)[-12:]} running status={snapshot.status} url={snapshot.url}", flush=True) + _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot) run_crawl( str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], selected_plugins=queued_plugins_for_snapshot(str(snapshot.id)), process_discovered_snapshots_inline=True, + interactive_interrupts=interactive_interrupts, ) return True @@ -1488,7 +1548,13 @@ def run_install(*, plugin_names: list[str] | None = None) -> None: asyncio.run(_run_install(plugin_names=plugin_names)) -def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, maintenance_only: bool = False) -> int: +def run_pending_crawls( + *, + daemon: bool = False, + crawl_id: str | None = None, + maintenance_only: bool = False, + interactive_interrupts: bool = False, +) -> int: from archivebox.crawls.models import Crawl, CrawlSchedule from archivebox.core.models import ArchiveResult, Snapshot from archivebox.machine.models import Binary, Process @@ -1515,7 +1581,11 @@ def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, mai due_crawls = due_crawls.filter(id=crawl_id) due_crawl = due_crawls.order_by("retry_at", "created_at").first() if due_crawl is not None: - if not run_due_crawl(due_crawl, lock_seconds=crawl_claim_lock_seconds): + if not run_due_crawl( + due_crawl, + lock_seconds=crawl_claim_lock_seconds, + interactive_interrupts=interactive_interrupts, + ): continue continue @@ -1526,7 +1596,7 @@ def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, mai due_snapshots = due_snapshots.filter(crawl_id=crawl_id) due_snapshot = due_snapshots.order_by("retry_at", "created_at").first() if due_snapshot is not None: - if not run_due_snapshot(due_snapshot, lock_seconds=60): + if not run_due_snapshot(due_snapshot, lock_seconds=60, interactive_interrupts=interactive_interrupts): continue continue diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index e48b8188..a6543212 100644 --- a/archivebox/services/snapshot_service.py +++ b/archivebox/services/snapshot_service.py @@ -1,7 +1,11 @@ from __future__ import annotations +import sys + from asgiref.sync import sync_to_async +from django.core.exceptions import ValidationError from django.utils import timezone +from rich import print as rprint from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent from abx_dl.limits import CrawlLimitState from abx_dl.services.base import BaseService @@ -27,7 +31,21 @@ class SnapshotService(BaseService): if snapshot.is_paused: return if snapshot.status == Snapshot.StatusChoices.QUEUED: - await sync_to_async(snapshot.sm.tick, thread_sensitive=True)() + try: + await sync_to_async(snapshot.sm.tick, thread_sensitive=True)() + except ValidationError as err: + if "ArchiveBox cannot archive its own admin, web, api, or snapshot URLs." not in str(err): + raise + await Snapshot.objects.filter(id=snapshot.id).aupdate( + status=Snapshot.StatusChoices.SEALED, + retry_at=None, + modified_at=timezone.now(), + ) + rprint( + f"[red][X] Refusing to archive ArchiveBox internal URL for security: {snapshot.url}[/red]", + file=sys.stderr, + ) + return await sync_to_async(snapshot.refresh_from_db, thread_sensitive=True)() elif snapshot.status != Snapshot.StatusChoices.STARTED: return diff --git a/archivebox/services/supervision_service.py b/archivebox/services/supervision_service.py index 93ee2086..d1280466 100644 --- a/archivebox/services/supervision_service.py +++ b/archivebox/services/supervision_service.py @@ -51,7 +51,7 @@ def command_is_newest(command, *, process_type: str, data_dir: str | Path, url: return bool(leader and leader.id == command.id) -def runtime_stack_owner(*, data_dir: str | Path): +def runtime_stack_owner(*, data_dir: str | Path, exclude_id=None): from archivebox.machine.models import Machine, Process base_qs = Process.objects.filter( @@ -60,6 +60,8 @@ def runtime_stack_owner(*, data_dir: str | Path): pwd=str(data_dir), process_type__in=runtime_stack_owner_types(), ) + if exclude_id is not None: + base_qs = base_qs.exclude(id=exclude_id) for process_types in ( (Process.TypeChoices.UPDATE,), (Process.TypeChoices.SERVER, Process.TypeChoices.ADD), @@ -141,28 +143,31 @@ def standby_until_leader_needed(command, *, process_type: str, data_dir: str | P if not announced: leader = newest_live_process(process_type=process_type, data_dir=data_dir, url=url) leader_pid = leader.pid if leader else "unknown" - rprint(f"[yellow][*] Standing by; newer ArchiveBox parent pid={leader_pid} is running the orchestrator and server.[/yellow]") + rprint(f"[yellow][*] Standing by; newer ArchiveBox process pid={leader_pid} is running the orchestrator and server.[/yellow]") announced = True time.sleep(interval) command.modified_at = timezone.now() command.save(update_fields=["modified_at"]) -def standby_until_runtime_stack_needed(command, *, data_dir: str | Path, interval: float = 2.0) -> None: +def standby_until_runtime_stack_needed(command, *, data_dir: str | Path, interval: float = 2.0) -> dict[str, object]: from archivebox.workers.supervisord_util import reap_foreground_supervisord_process announced = False + previous_owner_pid = None while not command_owns_runtime_stack(command, data_dir=data_dir): reap_foreground_supervisord_process() if not announced: owner = runtime_stack_owner(data_dir=data_dir) owner_pid = owner.pid if owner else "unknown" + previous_owner_pid = owner_pid rprint( - "[yellow][*] A newer archivebox process took over the runner " - f"(pid={owner_pid}). Work will continue there, and will continue here if the other process is stopped and work still remains.[/yellow]", + "[yellow][*] A newer archivebox process took over the runtime stack " + f"(pid={owner_pid}). Work will continue there, and will resume here if that process exits and work still remains.[/yellow]", file=sys.stderr, ) announced = True time.sleep(interval) command.modified_at = timezone.now() command.save(update_fields=["modified_at"]) + return {"resumed": announced, "previous_owner_pid": previous_owner_pid} diff --git a/archivebox/tests/test_archive_result_service.py b/archivebox/tests/test_archive_result_service.py index e5071edb..2c9a6e11 100644 --- a/archivebox/tests/test_archive_result_service.py +++ b/archivebox/tests/test_archive_result_service.py @@ -70,6 +70,81 @@ def test_process_completed_projects_inline_archiveresult(): _cleanup_machine_process_rows() +def test_archiveresult_event_retry_updates_existing_hook_row(): + from archivebox.core.models import ArchiveResult + from archivebox.services.archive_result_service import ArchiveResultService + import asyncio + + snapshot = _create_snapshot() + plugin_dir = Path(snapshot.output_dir) / "wget" + plugin_dir.mkdir(parents=True, exist_ok=True) + (plugin_dir / "index.html").write_text("ok") + + service = ArchiveResultService(create_bus(name="test_archiveresult_retry_updates_existing_hook_row")) + first_event = ArchiveResultEvent( + snapshot_id=str(snapshot.id), + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + status="failed", + output_str="timed out", + start_ts="2026-03-22T12:00:00+00:00", + end_ts="2026-03-22T12:00:01+00:00", + ) + retry_event = ArchiveResultEvent( + snapshot_id=str(snapshot.id), + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + status="succeeded", + output_str="wget/index.html", + output_files=[OutputFile(path="index.html", extension="html", mimetype="text/html", size=15)], + start_ts="2026-03-22T12:01:00+00:00", + end_ts="2026-03-22T12:01:01+00:00", + ) + + async def emit_events() -> None: + await service.on_ArchiveResultEvent__save_to_db(first_event) + first_result_id = await ArchiveResult.objects.values_list("id", flat=True).aget( + snapshot=snapshot, + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + ) + await service.on_ArchiveResultEvent__save_to_db(retry_event) + retry_result = await ArchiveResult.objects.aget( + snapshot=snapshot, + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + ) + assert retry_result.id == first_result_id + assert retry_result.status == ArchiveResult.StatusChoices.SUCCEEDED + assert retry_result.output_str == "wget/index.html" + + asyncio.run(emit_events()) + + assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="wget", hook_name="on_Snapshot__06_wget.finite.bg").count() == 1 + _cleanup_machine_process_rows() + + +def test_archiveresult_duplicate_hook_rows_are_rejected(): + from django.db import IntegrityError, transaction + from archivebox.core.models import ArchiveResult + + snapshot = _create_snapshot() + ArchiveResult.objects.create( + snapshot=snapshot, + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + status=ArchiveResult.StatusChoices.FAILED, + ) + + with pytest.raises(IntegrityError), transaction.atomic(): + ArchiveResult.objects.create( + snapshot=snapshot, + plugin="wget", + hook_name="on_Snapshot__06_wget.finite.bg", + status=ArchiveResult.StatusChoices.SUCCEEDED, + ) + + def test_process_completed_projects_synthetic_failed_archiveresult(): from archivebox.core.models import ArchiveResult from archivebox.services.archive_result_service import ArchiveResultService diff --git a/archivebox/tests/test_cli_real_flows.py b/archivebox/tests/test_cli_real_flows.py index 4f1cc6f9..2f0bc420 100644 --- a/archivebox/tests/test_cli_real_flows.py +++ b/archivebox/tests/test_cli_real_flows.py @@ -11,6 +11,7 @@ import sys import time from pathlib import Path +import psutil import pytest from archivebox.core.models import ArchiveResult, Snapshot @@ -145,6 +146,24 @@ def _wait_for_pid_to_disappear(pid: int, *, timeout: float = 20.0) -> None: raise AssertionError(f"PID {pid} is still running") +def _wait_for_process(predicate, *, timeout: float = 20.0): + deadline = time.time() + timeout + last_seen = [] + while time.time() < deadline: + last_seen = [] + for proc in psutil.process_iter(["pid", "ppid", "cmdline"]): + try: + cmdline = proc.info.get("cmdline") or [] + command = " ".join(cmdline) + last_seen.append(f"{proc.info.get('pid')} {proc.info.get('ppid')} {command}") + if predicate(proc, command): + return proc + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + time.sleep(0.2) + raise AssertionError("No matching live process found. Last seen:\n" + "\n".join(last_seen[-50:])) + + def _supervisor_pid_from_log(log_path: Path) -> int: content = log_path.read_text(encoding="utf-8", errors="replace") matches = re.findall(r"Supervisord connected \(pid=(\d+)\)", content) @@ -161,7 +180,37 @@ def _worker_pid_from_log(log_path: Path, worker_name: str) -> int: def _pgrep_data_dir(data_dir) -> list[str]: result = subprocess.run(["pgrep", "-af", str(data_dir)], capture_output=True, text=True, timeout=5) - return [line for line in result.stdout.splitlines() if "pgrep -af" not in line] + lines = [line for line in result.stdout.splitlines() if "pgrep -af" not in line] + + # A foreground ArchiveBox process can be killed with SIGKILL before Python + # cleanup runs. Supervisord's command line only points at its generated + # config file, so catch orphaned supervisors by resolving pidfiles whose + # configs still reference this real test DATA_DIR. + for runtime_root in (Path("/tmp/archivebox"), Path(data_dir) / "tmp"): + for config_path in runtime_root.glob("*/supervisord.conf"): + try: + config_text = config_path.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + if str(data_dir) not in config_text: + continue + pid_path = config_path.with_name("supervisord.pid") + try: + pid = int(pid_path.read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + continue + if not _pid_is_alive(pid): + continue + ps_line = subprocess.run( + ["ps", "-p", str(pid), "-o", "pid=,ppid=,command="], + capture_output=True, + text=True, + timeout=5, + ).stdout.strip() + if ps_line: + lines.append(ps_line) + + return sorted(set(lines)) def _assert_no_processes_for_data_dir(data_dir, *, timeout: float = 10.0) -> None: @@ -499,8 +548,47 @@ def test_live_server_signal_exit_and_resume_uses_existing_supervisor_state(tmp_p _kill_processes_for_data_dir(tmp_path) +@pytest.mark.timeout(180) +def test_live_daemonized_server_keeps_supervisord_owned_by_archivebox_parent(tmp_path, process): + os.chdir(tmp_path) + assert process.returncode == 0, process.stderr + + env = _live_exit_env(tmp_path) + port = _free_port() + bind_url = f"http://127.0.0.1:{port}" + try: + result = subprocess.run( + [sys.executable, "-m", "archivebox", "server", "--daemonize", f"127.0.0.1:{port}"], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + timeout=90, + ) + assert result.returncode == 0, result.stderr or result.stdout + _wait_for_port("127.0.0.1", port, timeout=30) + + server_process = _wait_for_process( + lambda _proc, command: "archivebox" in command and " server " in f" {command} " and bind_url.replace("http://", "") in command, + ) + supervisord = _wait_for_process( + lambda proc, command: proc.ppid() == server_process.pid and "supervisord" in command, + ) + _wait_for_process( + lambda proc, command: proc.ppid() == supervisord.pid and "supervisord_watchdog" in command, + ) + + os.kill(server_process.pid, signal.SIGKILL) + _wait_for_pid_to_disappear(server_process.pid, timeout=10) + _wait_for_pid_to_disappear(supervisord.pid, timeout=20) + _assert_no_processes_for_data_dir(tmp_path, timeout=12) + finally: + _kill_processes_for_data_dir(tmp_path) + _assert_no_processes_for_data_dir(tmp_path, timeout=12) + + @pytest.mark.timeout(240) -def test_live_second_server_takes_over_existing_server_parent(tmp_path, process): +def test_live_second_server_takes_over_existing_server_process(tmp_path, process): os.chdir(tmp_path) assert process.returncode == 0, process.stderr @@ -515,8 +603,8 @@ def test_live_second_server_takes_over_existing_server_parent(tmp_path, process) assert first.poll() is None first_text = first_log.read_text(encoding="utf-8", errors="replace") second_text = second_log.read_text(encoding="utf-8", errors="replace") - assert "Newer ArchiveBox server parent took over; standing by." in first_text - assert "is now running the orchestrator and server" in second_text + assert "A newer archivebox process took over the runtime stack" in first_text + assert "Starting orchestrator, server" in second_text status = subprocess.run( [sys.executable, "-m", "archivebox", "status"], @@ -528,10 +616,10 @@ def test_live_second_server_takes_over_existing_server_parent(tmp_path, process) ) assert status.returncode == 0, status.stderr or status.stdout - first_takeovers = first_log.read_text(encoding="utf-8", errors="replace").count("is now running the orchestrator and server") + first_resumes = first_log.read_text(encoding="utf-8", errors="replace").count("Other newer archivebox process") _stop_process(second, signal.SIGTERM) second = None - _wait_for_log_count(first_log, "is now running the orchestrator and server", first_takeovers + 1, timeout=35) + _wait_for_log_count(first_log, "Other newer archivebox process", first_resumes + 1, timeout=35) assert first.poll() is None finally: if second is not None and second.poll() is None: @@ -567,8 +655,8 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, process): current_log = log_path.read_text(encoding="utf-8", errors="replace") assert previous_server.poll() is None assert _pid_is_alive(server_pids[index - 1]) - assert "Newer ArchiveBox server parent took over; standing by." in previous_log - assert "is now running the orchestrator and server" in current_log + assert "A newer archivebox process took over the runtime stack" in previous_log + assert "Starting orchestrator, server" in current_log _wait_for_pid_to_disappear(daphne_pids[index - 1], timeout=15) _wait_for_pid_to_disappear(runner_pids[index - 1], timeout=15) @@ -596,10 +684,10 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, process): previous_log_path = tmp_path / "server-chaos-3.log" previous_takeovers = previous_log_path.read_text(encoding="utf-8", errors="replace").count( - "is now running the orchestrator and server", + "Other newer archivebox process", ) _stop_process(servers[-1], signal.SIGTERM) - _wait_for_log_count(previous_log_path, "is now running the orchestrator and server", previous_takeovers + 1, timeout=35) + _wait_for_log_count(previous_log_path, "Other newer archivebox process", previous_takeovers + 1, timeout=35) assert servers[3].poll() is None finally: for server in reversed(servers): diff --git a/archivebox/tests/test_cli_update_reindex_snapshots.py b/archivebox/tests/test_cli_update_reindex_snapshots.py index 5d68843a..d99c81aa 100644 --- a/archivebox/tests/test_cli_update_reindex_snapshots.py +++ b/archivebox/tests/test_cli_update_reindex_snapshots.py @@ -290,6 +290,62 @@ def test_reconcile_with_index_json_imports_legacy_archive_results_and_process(tm assert '"type": "Process"' in jsonl_text +@pytest.mark.django_db +def test_reconcile_with_index_json_merges_retried_archive_results(tmp_path): + from archivebox.base_models.models import get_or_create_system_user_pk + from archivebox.core.models import ArchiveResult, Snapshot + from archivebox.crawls.models import Crawl + + crawl = Crawl.objects.create( + urls="https://example.com", + created_by_id=get_or_create_system_user_pk(), + ) + snapshot = Snapshot.objects.create( + url="https://example.com", + crawl=crawl, + status=Snapshot.StatusChoices.SEALED, + ) + output_dir = snapshot.output_dir + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "index.json").write_text( + json.dumps( + { + "url": snapshot.url, + "timestamp": snapshot.timestamp, + "title": "Example Domain", + "archive_results": [ + { + "plugin": "dom", + "hook_name": "on_Snapshot__12_dom.js", + "status": "failed", + "output": "first attempt failed", + "start_ts": "2024-01-01T00:00:00+00:00", + "end_ts": "2024-01-01T00:00:01+00:00", + }, + { + "plugin": "dom", + "hook_name": "on_Snapshot__12_dom.js", + "status": "succeeded", + "output": "dom/output.html", + "output_files": {"output.html": {"size": 42}}, + "output_size": 42, + "start_ts": "2024-01-01T00:01:00+00:00", + "end_ts": "2024-01-01T00:01:01+00:00", + }, + ], + }, + ), + ) + + snapshot.reconcile_with_index_json() + + result = ArchiveResult.objects.get(snapshot=snapshot, plugin="dom", hook_name="on_Snapshot__12_dom.js") + assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="dom", hook_name="on_Snapshot__12_dom.js").count() == 1 + assert result.status == ArchiveResult.StatusChoices.SUCCEEDED + assert result.output_str == "dom/output.html" + assert result.output_size == 42 + + @pytest.mark.django_db def test_reconcile_with_index_json_trusts_legacy_archive_results(tmp_path): from archivebox.base_models.models import get_or_create_system_user_pk diff --git a/archivebox/tests/test_crawl_runner.py b/archivebox/tests/test_crawl_runner.py index 996d467e..8baff22b 100644 --- a/archivebox/tests/test_crawl_runner.py +++ b/archivebox/tests/test_crawl_runner.py @@ -420,38 +420,29 @@ def test_crawl_runner_empty_plugin_selection_emits_lifecycle_and_seals_crawl(tmp ) runner = CrawlRunner(crawl) + seen_events = { + CrawlEvent: [], + CrawlSetupEvent: [], + CrawlStartEvent: [], + SnapshotEvent: [], + SnapshotCompletedEvent: [], + CrawlCleanupEvent: [], + CrawlCompletedEvent: [], + MachineEvent: [], + } + for event_type, events in seen_events.items(): + runner.bus.on(event_type, lambda event, events=events: events.append(event)) + asyncio.run(runner.run()) - async def collect_events(): - crawl_events = await runner.bus.filter(CrawlEvent, past=True) - setup_events = await runner.bus.filter(CrawlSetupEvent, past=True) - start_events = await runner.bus.filter(CrawlStartEvent, past=True) - snapshot_events = await runner.bus.filter(SnapshotEvent, past=True) - snapshot_completed_events = await runner.bus.filter(SnapshotCompletedEvent, past=True) - cleanup_events = await runner.bus.filter(CrawlCleanupEvent, past=True) - completed_events = await runner.bus.filter(CrawlCompletedEvent, past=True) - machine_events = await runner.bus.filter(MachineEvent, past=True) - return ( - crawl_events, - setup_events, - start_events, - snapshot_events, - snapshot_completed_events, - cleanup_events, - completed_events, - machine_events, - ) - - ( - crawl_events, - setup_events, - start_events, - snapshot_events, - snapshot_completed_events, - cleanup_events, - completed_events, - machine_events, - ) = asyncio.run(collect_events()) + crawl_events = seen_events[CrawlEvent] + setup_events = seen_events[CrawlSetupEvent] + start_events = seen_events[CrawlStartEvent] + snapshot_events = seen_events[SnapshotEvent] + snapshot_completed_events = seen_events[SnapshotCompletedEvent] + cleanup_events = seen_events[CrawlCleanupEvent] + completed_events = seen_events[CrawlCompletedEvent] + machine_events = seen_events[MachineEvent] assert len(crawl_events) == 1 assert len(setup_events) == 1 @@ -474,6 +465,7 @@ def test_crawl_runner_empty_plugin_selection_emits_lifecycle_and_seals_crawl(tmp assert crawl.retry_at is None assert snapshot.status == Snapshot.StatusChoices.SEALED assert snapshot.retry_at is None + assert snapshot.archiveresult_set.count() == 0 @pytest.mark.django_db(transaction=True) @@ -789,6 +781,7 @@ def test_create_crawl_api_queues_crawl_without_spawning_runner(): assert str(crawl.id) assert crawl.status == "queued" assert crawl.retry_at is not None + assert crawl.snapshot_set.filter(url="https://example.com").count() == 1 def test_wait_for_snapshot_tasks_surfaces_already_failed_task(): diff --git a/archivebox/workers/management/commands/runner_watch.py b/archivebox/workers/management/commands/runner_watch.py index af5f5e57..470f252e 100644 --- a/archivebox/workers/management/commands/runner_watch.py +++ b/archivebox/workers/management/commands/runner_watch.py @@ -25,7 +25,7 @@ class Command(BaseCommand): from archivebox.machine.models import Machine, Process from archivebox.workers.supervisord_util import ( RUNNER_WORKER, - get_existing_supervisord_process, + SupervisordConnectionCache, get_worker, start_worker, stop_worker, @@ -43,6 +43,7 @@ class Command(BaseCommand): interval = max(0.2, float(kwargs.get("interval", 1.0))) last_runserver_id = None + supervisor_cache = SupervisordConnectionCache() def stop_duplicate_watchers() -> None: machine = Machine.current() @@ -58,7 +59,7 @@ class Command(BaseCommand): proc.terminate(graceful_timeout=2.0) def get_supervisor(): - supervisor = get_existing_supervisord_process() + supervisor = supervisor_cache.get() if supervisor is None: raise RuntimeError("runner_watch requires a running supervisord process") return supervisor @@ -114,6 +115,7 @@ class Command(BaseCommand): restart_runner() current.heartbeat() except Exception: + supervisor_cache.clear() pass time.sleep(interval) diff --git a/archivebox/workers/management/commands/supervisord_watchdog.py b/archivebox/workers/management/commands/supervisord_watchdog.py new file mode 100644 index 00000000..cbd50280 --- /dev/null +++ b/archivebox/workers/management/commands/supervisord_watchdog.py @@ -0,0 +1,54 @@ +import time + +import psutil +from django.core.management.base import BaseCommand + + +class Command(BaseCommand): + help = "Stop a foreground-owned supervisord if its exact ArchiveBox owner Process exits." + + def add_arguments(self, parser): + parser.add_argument("--supervisord-process-id", required=True) + parser.add_argument("--interval", type=float, default=1.0) + + def handle(self, *args, **kwargs): + from archivebox.core.shutdown_util import wait_psutil_and_kill_children + from archivebox.machine.models import Process + + supervisord_process_id = kwargs["supervisord_process_id"] + interval = max(0.2, float(kwargs["interval"])) + + while True: + try: + supervisord_process = Process.objects.select_related("parent").get(id=supervisord_process_id) + except Process.DoesNotExist: + return + + if supervisord_process.status != Process.StatusChoices.RUNNING: + return + + supervisord = supervisord_process.proc + if supervisord is None: + supervisord_process.mark_exited(exit_code=0) + return + + owner = supervisord_process.parent + if owner is not None and owner.is_running: + time.sleep(interval) + continue + + try: + children = supervisord.children(recursive=True) + supervisord.terminate() + for child in children: + try: + child.terminate() + except psutil.NoSuchProcess: + pass + wait_psutil_and_kill_children(supervisord, children, timeout=5) + supervisord_process.mark_exited(exit_code=0) + except psutil.NoSuchProcess: + supervisord_process.mark_exited(exit_code=0) + except (BrokenPipeError, OSError, psutil.TimeoutExpired): + pass + return diff --git a/archivebox/workers/supervisord_util.py b/archivebox/workers/supervisord_util.py index 68ca6f38..cd93d796 100644 --- a/archivebox/workers/supervisord_util.py +++ b/archivebox/workers/supervisord_util.py @@ -8,6 +8,7 @@ import psutil import shutil import subprocess import shlex +import signal from typing import cast from pathlib import Path @@ -37,6 +38,8 @@ WORKERS_DIR_NAME = "workers" # Global reference to supervisord process for cleanup _supervisord_proc = None _desired_supervisord_workers: dict[str, dict[str, str]] = {} +_ACTIVE_WORKER_STATES = {"STARTING", "RUNNING", "BACKOFF"} +_RUNTIME_COMPONENT_ORDER = ("orchestrator", "server", "sonic") def _shell_join(args: list[str]) -> str: @@ -182,6 +185,27 @@ RUNNER_WATCH_WORKER = lambda bind_url: { "redirect_stderr": "true", } +SUPERVISORD_PARENT_WATCHDOG_WORKER = lambda supervisord_process_id: { + "name": "worker_supervisord_parent_watchdog", + "command": _shell_join( + [ + sys.executable, + "-m", + "archivebox", + "manage", + "supervisord_watchdog", + f"--supervisord-process-id={supervisord_process_id}", + ], + ), + "autostart": "false", + "autorestart": "false", + "stopasgroup": "true", + "killasgroup": "true", + "stopwaitsecs": "1", + "stdout_logfile": "logs/worker_supervisord_parent_watchdog.log", + "redirect_stderr": "true", +} + SERVER_WORKER = lambda host, port: { "name": "worker_daphne", "command": _shell_join( @@ -191,7 +215,7 @@ SERVER_WORKER = lambda host, port: { "daphne", f"--bind={host}", f"--port={port}", - "--application-close-timeout=15", + "--application-close-timeout=0", "archivebox.core.asgi:application", ], ), @@ -199,7 +223,7 @@ SERVER_WORKER = lambda host, port: { "autorestart": "true", "stopasgroup": "true", "killasgroup": "true", - "stopwaitsecs": "20", + "stopwaitsecs": "1", "stdout_logfile": "logs/worker_daphne.log", "redirect_stderr": "true", } @@ -227,6 +251,9 @@ def RUNSERVER_WORKER(host: str, port: str, *, reload: bool, nothreading: bool = "environment": ",".join(environment), "autostart": "false", "autorestart": "true", + "stopasgroup": "true", + "killasgroup": "true", + "stopwaitsecs": "1", "stdout_logfile": "logs/worker_runserver.log", "redirect_stderr": "true", } @@ -316,6 +343,29 @@ def create_worker_config(daemon): worker_conf.write_text(worker_str) +def _current_foreground_supervisord_process_id(): + if not _supervisord_proc or _supervisord_proc.poll() is not None: + return None + + try: + from archivebox.machine.models import Machine, Process + + current = Process.current() + for process in Process.objects.filter( + machine=Machine.current(), + process_type=Process.TypeChoices.SUPERVISORD, + status=Process.StatusChoices.RUNNING, + pwd=str(CONSTANTS.DATA_DIR), + pid=_supervisord_proc.pid, + parent=current, + ).iterator(chunk_size=10): + if process.is_running: + return process.id + except Exception: + return None + return None + + def sync_supervisord_workers(supervisor, workers: list[tuple[dict[str, str], bool]], *, prune: bool = True): """Project desired workers into supervisord from ArchiveBox-owned state. @@ -331,6 +381,12 @@ def sync_supervisord_workers(supervisor, workers: list[tuple[dict[str, str], boo global _desired_supervisord_workers + supervisord_process_id = _current_foreground_supervisord_process_id() + if supervisord_process_id is not None: + watchdog = SUPERVISORD_PARENT_WATCHDOG_WORKER(supervisord_process_id) + if all(worker["name"] != watchdog["name"] for worker, _lazy in workers): + workers = [*workers, (watchdog, False)] + desired = {worker["name"]: (worker, lazy) for worker, lazy in workers} if prune: _desired_supervisord_workers = {name: worker for name, (worker, _lazy) in desired.items()} @@ -416,11 +472,62 @@ def get_existing_supervisord_process(*, quiet: bool = False): return None +class SupervisordConnectionCache: + """Reuse one XML-RPC proxy until it fails, avoiding hot-loop reconnects.""" + + def __init__(self, *, quiet: bool = False): + self.quiet = quiet + self.supervisor = None + + def clear(self) -> None: + self.supervisor = None + + def get(self): + if self.supervisor is not None: + try: + self.supervisor.getPID() + return self.supervisor + except Exception: + self.supervisor = None + + supervisor = get_existing_supervisord_process(quiet=self.quiet) + if supervisor is None: + return None + + self.supervisor = supervisor + return supervisor + + def stop_existing_supervisord_process(): global _supervisord_proc SOCK_FILE = get_sock_file() PID_FILE = SOCK_FILE.parent / PID_FILE_NAME stop_grace_seconds = configured_stopwaitsecs(tuple(_desired_supervisord_workers.values())) + live_supervisord = _live_supervisord_processes_from_db() + + for process, _proc in live_supervisord: + if process is None or process.parent_id is None: + continue + owner = process.parent + if owner.pid == os.getpid() or not owner.is_running: + continue + owner_proc = owner.proc + if owner_proc is None: + owner.mark_exited(exit_code=0) + continue + try: + print(f"[πŸ¦Έβ€β™‚οΈ] Stopping older ArchiveBox runtime owner (pid={owner_proc.pid})...") + owner_proc.terminate() + try: + owner_proc.wait(timeout=min(stop_grace_seconds, 5)) + except psutil.TimeoutExpired: + owner_proc.kill() + owner_proc.wait(timeout=2) + owner.mark_exited(exit_code=0) + except psutil.NoSuchProcess: + owner.mark_exited(exit_code=0) + except (BrokenPipeError, OSError, psutil.TimeoutExpired): + pass supervisor = get_existing_supervisord_process(quiet=True) supervisor_pid = None @@ -691,7 +798,7 @@ def start_worker(supervisor, daemon, lazy=False): return sync_supervisord_workers(supervisor, [(daemon, lazy)], prune=False).get(daemon["name"]) -def run_runner_worker(args: list[str], *, name: str = "worker_runner_once") -> int: +def run_runner_worker(args: list[str], *, name: str = "worker_runner_once", interactive_interrupts: bool = False) -> int: supervisor = get_or_create_supervisord_process(daemonize=False) worker = RUNNER_ONCE_WORKER(args, name=name) log_path = Path(worker["stdout_logfile"]) @@ -703,26 +810,38 @@ def run_runner_worker(args: list[str], *, name: str = "worker_runner_once") -> i log_handle.seek(0, 2) sync_supervisord_workers(supervisor, [(worker, False)], prune=False) final_states = {"STOPPED", "EXITED", "FATAL", "UNKNOWN"} + forwarded_interrupt = False try: while True: - while True: - line = log_handle.readline() - if not line: - break - sys.stdout.write(line) - sys.stdout.flush() - proc = get_worker(supervisor, name) - if proc is None: - return 1 - if proc["statename"] in final_states: + try: while True: line = log_handle.readline() if not line: break sys.stdout.write(line) sys.stdout.flush() - return int(proc.get("exitstatus") or 0) if proc["statename"] == "EXITED" else 1 - time.sleep(0.5) + proc = get_worker(supervisor, name) + if proc is None: + return 1 + if proc["statename"] in final_states: + while True: + line = log_handle.readline() + if not line: + break + sys.stdout.write(line) + sys.stdout.flush() + return int(proc.get("exitstatus") or 0) if proc["statename"] == "EXITED" else 1 + time.sleep(0.5) + except KeyboardInterrupt: + if not interactive_interrupts or forwarded_interrupt: + raise + forwarded_interrupt = True + proc = get_worker(supervisor, name) + pid = int(proc.get("pid") or 0) if proc else 0 + if pid <= 0: + raise + print("[yellow][*] Forwarding Ctrl+C to the active crawl hook...[/yellow]") + os.kill(pid, signal.SIGINT) finally: log_handle.close() @@ -735,6 +854,82 @@ def get_worker(supervisor, daemon_name): return None +def format_runtime_components(components: list[str] | tuple[str, ...]) -> str: + return ", ".join(component for component in components if component) + + +def worker_runtime_component(worker_name: str, *, config=None) -> str | None: + if worker_name in {RUNNER_WORKER["name"], RUNNER_WATCH_WORKER("")["name"]}: + return "orchestrator" + if worker_name in {SERVER_WORKER("", "")["name"], RUNSERVER_WORKER("", "", reload=False)["name"]}: + return "server" + if config is not None: + sonic_worker = get_sonic_supervisord_worker_from_plugin(config) + if sonic_worker and worker_name == sonic_worker.get("name"): + return "sonic" + return None + + +def runtime_components_for_worker_names(worker_names: set[str] | list[str] | tuple[str, ...], *, config=None) -> list[str]: + components = {worker_runtime_component(worker_name, config=config) for worker_name in worker_names} + return [component for component in _RUNTIME_COMPONENT_ORDER if component in components] + + +def active_supervisord_runtime_components(*, config=None, supervisor=None) -> list[str]: + supervisor = supervisor or get_existing_supervisord_process(quiet=True) + if supervisor is None: + return [] + try: + worker_names = {proc.get("name") for proc in supervisor.getAllProcessInfo() if proc.get("statename") in _ACTIVE_WORKER_STATES} + except Exception: + return [] + return runtime_components_for_worker_names({str(name) for name in worker_names if name}, config=config) + + +def build_server_worker_plan(*, config, host: str, port: str, debug: bool, reload: bool, nothreading: bool, supervisor=None): + bind_url = f"http://{host}:{port}" + + if debug: + server_worker = RUNSERVER_WORKER(host=host, port=port, reload=reload, nothreading=nothreading) + bg_workers: list[tuple[dict[str, str], bool]] = ( + [(RUNNER_WORKER, True), (RUNNER_WATCH_WORKER(bind_url), False)] if reload else [(RUNNER_WORKER, False)] + ) + log_files = ["logs/worker_runserver.log", "logs/worker_runner.log"] + if reload: + log_files.insert(1, "logs/worker_runner_watch.log") + else: + server_worker = SERVER_WORKER(host=host, port=port) + bg_workers = [(RUNNER_WORKER, False)] + log_files = ["logs/worker_daphne.log", "logs/worker_runner.log"] + + sonic_worker = get_sonic_supervisord_worker_from_plugin(config) + if sonic_worker is not None: + try: + current_sonic = get_worker(supervisor, sonic_worker["name"]) if supervisor is not None else None + supervisor_pid = supervisor.getPID() if supervisor is not None else None + except Exception: + current_sonic = None + supervisor_pid = None + sonic_host = str(getattr(config, "SEARCH_BACKEND_SONIC_HOST_NAME", "127.0.0.1") or "127.0.0.1") + if sonic_host.strip().lower() == "localhost": + sonic_host = "127.0.0.1" + sonic_port = int(getattr(config, "SEARCH_BACKEND_SONIC_PORT")) + if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")): + stop_stale_sonic_processes(sonic_worker, supervisor_pid=supervisor_pid, host=sonic_host, port=sonic_port) + if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")) and is_port_in_use( + sonic_host, + sonic_port, + ): + print(f"[yellow][*] Sonic is already listening on {sonic_host}:{sonic_port}; not starting a duplicate worker.[/yellow]") + else: + bg_workers.insert(0, (sonic_worker, False)) + log_files.append(str(sonic_worker["stdout_logfile"])) + + workers = [(server_worker, False), *bg_workers] + components = runtime_components_for_worker_names([worker["name"] for worker, _lazy in workers], config=config) + return workers, log_files, components + + def stop_worker(supervisor, daemon_name): proc = get_worker(supervisor, daemon_name) @@ -806,7 +1001,7 @@ def tail_multiple_worker_logs(log_files: list[str], follow=True, proc=None, keep try: while follow: if keep_running is not None and not keep_running(): - print("\n[newer ArchiveBox parent is now running the orchestrator and server]") + print("\n[newer ArchiveBox process is now running the orchestrator and server]") return "transferred" # Check if the monitored process has exited @@ -857,11 +1052,87 @@ def get_sonic_supervisord_worker_from_plugin(config) -> dict[str, str] | None: return cast(dict[str, str] | None, worker) -def stop_stale_sonic_processes(sonic_worker: dict[str, str], *, supervisor_pid: int | None) -> None: +def _proc_cmdline(proc: psutil.Process) -> list[str]: + try: + return proc.cmdline() + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + return [] + + +def _is_sonic_process(proc: psutil.Process) -> bool: + cmdline = _proc_cmdline(proc) + return bool(cmdline and Path(cmdline[0]).name == "sonic") + + +def _is_supervisord_process(proc: psutil.Process | None) -> bool: + if proc is None: + return False + cmdline = _proc_cmdline(proc) + return any(Path(part).name == "supervisord" for part in cmdline) + + +def _has_live_archivebox_parent(proc: psutil.Process | None) -> bool: + try: + parent = proc.parent() if proc else None + except (psutil.NoSuchProcess, psutil.AccessDenied): + return False + if parent is None or parent.pid <= 1: + return False + cmdline = _proc_cmdline(parent) + return any("archivebox" in part for part in cmdline) + + +def _terminate_process_tree(root: psutil.Process, *, timeout: float = 2.0) -> None: + try: + children = root.children(recursive=True) + except psutil.NoSuchProcess: + return + try: + root.terminate() + except psutil.NoSuchProcess: + return + for child in children: + try: + child.terminate() + except psutil.NoSuchProcess: + pass + _gone, alive = psutil.wait_procs([root, *children], timeout=timeout) + for proc in alive: + try: + proc.kill() + except psutil.NoSuchProcess: + pass + psutil.wait_procs(alive, timeout=timeout) + + +def _sonic_listeners(host: str, port: int) -> list[psutil.Process]: + listeners = [] + for proc in psutil.process_iter(["pid", "name", "cmdline"]): + if not _is_sonic_process(proc): + continue + try: + connections = proc.net_connections(kind="tcp") + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + for conn in connections: + if conn.status != psutil.CONN_LISTEN or not conn.laddr or conn.laddr.port != port: + continue + addr = str(conn.laddr.ip) + if host in {"0.0.0.0", "::", addr} or addr in {"0.0.0.0", "::"}: + listeners.append(proc) + break + return listeners + + +def stop_stale_sonic_processes( + sonic_worker: dict[str, str], + *, + supervisor_pid: int | None, + host: str | None = None, + port: int | None = None, +) -> None: command = shlex.split(sonic_worker.get("command") or "") config_path = Path(command[command.index("-c") + 1]).resolve() if "-c" in command and command.index("-c") + 1 < len(command) else None - if config_path is None: - return stale = [] for proc in psutil.process_iter(["pid", "ppid", "name", "cmdline"]): @@ -869,28 +1140,37 @@ def stop_stale_sonic_processes(sonic_worker: dict[str, str], *, supervisor_pid: cmdline = proc.info.get("cmdline") or [] if proc.info["pid"] == os.getpid() or proc.info["ppid"] == supervisor_pid: continue - if Path(cmdline[0]).name != "sonic" or str(config_path) not in cmdline: + if config_path is None or Path(cmdline[0]).name != "sonic" or str(config_path) not in cmdline: continue stale.append(proc) except (IndexError, psutil.NoSuchProcess, psutil.AccessDenied): continue + if host is not None and port is not None: + for proc in _sonic_listeners(host, port): + try: + proc_ppid = proc.ppid() + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + if proc.pid == os.getpid() or proc_ppid == supervisor_pid: + continue + try: + supervisor = proc.parent() + except (psutil.NoSuchProcess, psutil.AccessDenied): + supervisor = None + if _is_supervisord_process(supervisor) and not _has_live_archivebox_parent(supervisor): + stale.append(supervisor) + elif proc_ppid <= 1: + stale.append(proc) + if not stale: return - print(f"[yellow][*] Taking over stale Sonic daemon(s) using {pretty_path(config_path)}...[/yellow]") - for proc in stale: - try: - proc.terminate() - except psutil.NoSuchProcess: - pass - _gone, alive = psutil.wait_procs(stale, timeout=2.0) - for proc in alive: - try: - proc.kill() - except psutil.NoSuchProcess: - pass - psutil.wait_procs(alive, timeout=2.0) + unique_stale = {proc.pid: proc for proc in stale}.values() + target = f"{host}:{port}" if host and port else pretty_path(config_path) if config_path else "unknown Sonic target" + print(f"[yellow][*] Taking over stale Sonic daemon(s) using {target}...[/yellow]") + for proc in unique_stale: + _terminate_process_tree(proc) def start_server_workers( @@ -902,6 +1182,7 @@ def start_server_workers( nothreading=False, keep_running=None, should_stop_supervisord=None, + resumed_from_pid=None, ): from archivebox.config.common import get_config @@ -910,46 +1191,26 @@ def start_server_workers( tail_result = "stopped" try: supervisor = get_or_create_supervisord_process(daemonize=daemonize) - bind_url = f"http://{host}:{port}" - - if debug: - server_worker = RUNSERVER_WORKER(host=host, port=port, reload=reload, nothreading=nothreading) - bg_workers: list[tuple[dict[str, str], bool]] = ( - [(RUNNER_WORKER, True), (RUNNER_WATCH_WORKER(bind_url), False)] if reload else [(RUNNER_WORKER, False)] + workers, log_files, components = build_server_worker_plan( + config=config, + host=host, + port=port, + debug=debug, + reload=reload, + nothreading=nothreading, + supervisor=supervisor, + ) + component_list = format_runtime_components(components) + if resumed_from_pid: + print( + "[yellow][*] Other newer archivebox process " + f"(pid={resumed_from_pid}) exited, taking over {component_list} in this process again...[/yellow]", ) - log_files = ["logs/worker_runserver.log", "logs/worker_runner.log"] - if reload: - log_files.insert(1, "logs/worker_runner_watch.log") else: - server_worker = SERVER_WORKER(host=host, port=port) - bg_workers = [(RUNNER_WORKER, False)] - log_files = ["logs/worker_daphne.log", "logs/worker_runner.log"] - - sonic_worker = get_sonic_supervisord_worker_from_plugin(config) - if sonic_worker is not None: - try: - current_sonic = get_worker(supervisor, sonic_worker["name"]) - supervisor_pid = supervisor.getPID() - except Exception: - current_sonic = None - supervisor_pid = None - if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")): - stop_stale_sonic_processes(sonic_worker, supervisor_pid=supervisor_pid) - sonic_host = str(getattr(config, "SEARCH_BACKEND_SONIC_HOST_NAME", "127.0.0.1") or "127.0.0.1") - if sonic_host.strip().lower() == "localhost": - sonic_host = "127.0.0.1" - sonic_port = int(getattr(config, "SEARCH_BACKEND_SONIC_PORT")) - if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")) and is_port_in_use( - sonic_host, - sonic_port, - ): - print(f"[yellow][*] Sonic is already listening on {sonic_host}:{sonic_port}; not starting a duplicate worker.[/yellow]") - else: - bg_workers.insert(0, (sonic_worker, False)) - log_files.append(str(sonic_worker["stdout_logfile"])) + print(f"[*] Starting {component_list} in this process (pid={os.getpid()})...") print() - sync_supervisord_workers(supervisor, [(server_worker, False), *bg_workers], prune=True) + sync_supervisord_workers(supervisor, workers, prune=True) print() if daemonize: @@ -987,7 +1248,7 @@ def start_server_workers( # Ensure supervisord and all children are stopped only while this # foreground parent is still the active server parent. Standby # parents must not tear down a newer leader's services. - stop_existing_supervisord_process() + stop_own_supervisord_process() return tail_result diff --git a/bin/release.sh b/bin/release.sh index 291f4214..1491cd35 100755 --- a/bin/release.sh +++ b/bin/release.sh @@ -221,6 +221,31 @@ else: PY } +github_release_enabled() { + local version="$1" + if [[ "${version}" == *rc* && "${CREATE_GITHUB_RC_RELEASES:-0}" != "1" ]]; then + return 1 + fi + if [[ "${CREATE_GITHUB_RELEASES:-1}" == "0" ]]; then + return 1 + fi + return 0 +} + +create_git_tag() { + local version="$1" + local tag="${TAG_PREFIX}${version}" + if git rev-parse -q --verify "refs/tags/${tag}" >/dev/null; then + if [[ "$(git rev-list -n1 "${tag}")" != "$(git rev-parse HEAD)" ]]; then + echo "Tag ${tag} already exists but does not point at HEAD" >&2 + return 1 + fi + else + git tag -a "${tag}" -m "release: ${tag}" + fi + git push origin "refs/tags/${tag}" +} + wait_for_runs() { local slug="$1" local event="$2" @@ -337,6 +362,10 @@ create_release() { local slug="$1" local version="$2" local prerelease_args=() + if ! github_release_enabled "${version}"; then + echo "Skipping GitHub release object for ${TAG_PREFIX}${version}; git tag will still be pushed." + return 0 + fi if [[ "${version}" == *rc* ]]; then prerelease_args+=(--prerelease) fi @@ -437,9 +466,10 @@ main() { fi publish_artifacts "${version}" + create_git_tag "${version}" create_release "${slug}" "${version}" - if ! gh release view "${TAG_PREFIX}${version}" --repo "${slug}" >/dev/null 2>&1; then + if github_release_enabled "${version}" && ! gh release view "${TAG_PREFIX}${version}" --repo "${slug}" >/dev/null 2>&1; then echo "GitHub release ${TAG_PREFIX}${version} was not found after creation" >&2 return 1 fi diff --git a/bin/release_dev_stack.sh b/bin/release_dev_stack.sh index afd46631..efd5fc50 100755 --- a/bin/release_dev_stack.sh +++ b/bin/release_dev_stack.sh @@ -133,6 +133,7 @@ commit_push_publish() { local branch="$2" local package="$3" local version="$4" + local tag="v${version}" ( cd "$repo" @@ -146,6 +147,15 @@ commit_push_publish() { echo "[*] No staged changes in ${package}; reusing existing commit." fi git push origin "$branch" + if git rev-parse -q --verify "refs/tags/${tag}" >/dev/null; then + if [[ "$(git rev-list -n1 "${tag}")" != "$(git rev-parse HEAD)" ]]; then + echo "[X] Tag ${tag} already exists but does not point at HEAD in ${package}" >&2 + exit 1 + fi + else + git tag -a "${tag}" -m "release: ${package} ${version}" + fi + git push origin "refs/tags/${tag}" uv --no-cache publish --username="${PYPI_USERNAME}" dist/* ) } diff --git a/etc/package.json b/etc/package.json index 8a6a14ca..7edb5299 100644 --- a/etc/package.json +++ b/etc/package.json @@ -1,6 +1,6 @@ { "name": "archivebox", - "version": "0.9.33rc26", + "version": "0.9.33rc29", "repository": "github:ArchiveBox/ArchiveBox", "license": "MIT", "dependencies": { diff --git a/pyproject.toml b/pyproject.toml index 479e0794..47bd8e1d 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "archivebox" -version = "0.9.33rc26" +version = "0.9.33rc29" requires-python = ">=3.13" description = "Self-hosted internet archiving solution." authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}] @@ -79,9 +79,9 @@ dependencies = [ ### Extractor dependencies (optional binaries detected at runtime via shutil.which) ### Binary/Package Management "abxbus==2.5.7", # EventBus API - "abxpkg>=1.11.51", # for: detecting, versioning, and installing binaries via apt/brew/pip/npm - "abx-plugins>=1.11.58", # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring - "abx-dl>=1.11.58", # shared ArchiveBox downloader package with blocking install preflight + "abxpkg>=1.11.53", # for: detecting, versioning, and installing binaries via apt/brew/pip/npm + "abx-plugins>=1.11.60", # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring + "abx-dl>=1.11.60", # shared ArchiveBox downloader package with blocking install preflight ### UUID7 backport for Python <3.14 "uuid7>=0.1.0; python_version < '3.14'", # provides the uuid_extensions module on Python 3.13 ]