diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py index 218abb10..e0d9e083 100644 --- a/archivebox/cli/archivebox_add.py +++ b/archivebox/cli/archivebox_add.py @@ -137,7 +137,7 @@ def add( persona_name = (persona or "Default").strip() or "Default" plugins = plugins or "" - persona_obj, _ = Persona.objects.get_or_create(name=persona_name) + persona_obj = Persona.get_or_create_named(persona_name) persona_obj.ensure_dirs() effective_persona_config = get_config(persona=persona_obj) diff --git a/archivebox/core/views.py b/archivebox/core/views.py index f7aa2101..f9a5c870 100644 --- a/archivebox/core/views.py +++ b/archivebox/core/views.py @@ -1503,6 +1503,10 @@ class AddView(UserPassesTestMixin, FormView): def form_valid(self, form): crawl = self._create_crawl_from_form(form) + if crawl.status in crawl.RUNNABLE_STATES: + from archivebox.services.runner import ensure_background_runner + + ensure_background_runner(allow_under_pytest=True) urls = form.cleaned_data["url"] schedule = form.cleaned_data.get("schedule", "").strip() diff --git a/archivebox/personas/models.py b/archivebox/personas/models.py index ba36a17f..d8f87696 100644 --- a/archivebox/personas/models.py +++ b/archivebox/personas/models.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any from collections.abc import Mapping -from django.db import models +from django.db import IntegrityError, models from django.db.models.fields.json import KT from django.conf import settings from django.utils import timezone @@ -228,6 +228,18 @@ class Persona(ModelWithConfig): if fcntl is not None: fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + @classmethod + def get_or_create_named(cls, name: str) -> "Persona": + persona_name = (name or "Default").strip() or "Default" + persona = cls.objects.filter(name=persona_name).first() + if persona is not None: + return persona + + try: + return cls.objects.create(name=persona_name) + except IntegrityError: + return cls.objects.get(name=persona_name) + def runtime_root_for_crawl(self, crawl) -> Path: return Path(crawl.output_dir) / ".persona" / self.name @@ -331,8 +343,7 @@ class Persona(ModelWithConfig): @classmethod def get_or_create_default(cls) -> "Persona": """Get or create the Default persona.""" - persona, _ = cls.objects.get_or_create(name="Default") - return persona + return cls.get_or_create_named("Default") @classmethod def cleanup_chrome_all(cls) -> int: diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 4b8218ad..c4ea6021 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -731,7 +731,7 @@ class CrawlRunner: crawl=snapshot.crawl, snapshot=snapshot, persona=self.persona, - runtime_overrides=runtime_chrome_overrides, + runtime_overrides={**runtime_chrome_overrides, **self.config_overrides}, extra_context={ "snapshot_id": str(snapshot.id), "snapshot_depth": snapshot.depth, @@ -1334,6 +1334,16 @@ def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: return queued_plugins +def _selected_plugin_config_overrides(selected_plugins: list[str] | None) -> dict[str, Any]: + config_overrides: dict[str, Any] = {} + if selected_plugins: + config_overrides["PLUGINS"] = ",".join(selected_plugins) + for plugin_name in selected_plugins or []: + if plugin_name.startswith("search_backend_"): + config_overrides[f"{plugin_name.upper()}_ENABLED"] = True + return config_overrides + + def fail_unavailable_queued_hooks( snapshot_id: str, selected_hooks_by_plugin: dict[str, set[str] | None], @@ -1588,13 +1598,26 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + config_overrides=_selected_plugin_config_overrides(selected_plugins), selected_plugins_are_explicit=False, ) finally: # Targeted plugin rows can complete while the Snapshot remains - # paused. Put retry_at back at MAX so the orchestrator leaves the - # paused lifecycle alone until an explicit resume transition. - snapshot.restore_paused_scheduler_marker() + # paused. Put retry_at back at MAX only after the queued rows are + # gone; if a hook was interrupted before projection, keep the + # paused row due so the next runner can retry that targeted work + # without a user-visible resume transition. + if queued_plugins_for_snapshot(str(snapshot.id)): + now = timezone.now() + type(snapshot).objects.filter( + pk=snapshot.pk, + status=snapshot.StatusChoices.PAUSED, + ).update( + retry_at=now, + modified_at=now, + ) + else: + snapshot.restore_paused_scheduler_marker() return True if snapshot.status == Snapshot.StatusChoices.SEALED: if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds): @@ -1620,6 +1643,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + config_overrides=_selected_plugin_config_overrides(selected_plugins), selected_plugins_are_explicit=False, ) if search_only_plugins: @@ -1637,6 +1661,14 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo retry_at=None, modified_at=timezone.now(), ) + else: + type(snapshot).objects.filter( + pk=snapshot.pk, + status=snapshot.StatusChoices.SEALED, + ).update( + retry_at=timezone.now(), + modified_at=timezone.now(), + ) return True if maintenance_ran: return True @@ -1675,12 +1707,14 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED") return True _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot) + selected_plugins = queued_plugins_for_snapshot(str(snapshot.id)) run_crawl( str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], - selected_plugins=queued_plugins_for_snapshot(str(snapshot.id)), + selected_plugins=selected_plugins, process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts, + config_overrides=_selected_plugin_config_overrides(selected_plugins), selected_plugins_are_explicit=False, ) snapshot.refresh_from_db() @@ -1955,12 +1989,8 @@ def _run_due_queued_plugin_result( if not claimed_snapshot_ids or selected_plugins is None: return True - config_overrides = { - "CRAWL_MAX_CONCURRENT_SNAPSHOTS": batch_size, - } - for plugin_name in selected_plugins: - if plugin_name.startswith("search_backend_"): - config_overrides[f"{plugin_name.upper()}_ENABLED"] = True + config_overrides = _selected_plugin_config_overrides(selected_plugins) + config_overrides["CRAWL_MAX_CONCURRENT_SNAPSHOTS"] = batch_size run_crawl( root_crawl_id, diff --git a/archivebox/tests/test_api_v1_cli_add.py b/archivebox/tests/test_api_v1_cli_add.py index b8623e63..8ca9d2bc 100644 --- a/archivebox/tests/test_api_v1_cli_add.py +++ b/archivebox/tests/test_api_v1_cli_add.py @@ -1,6 +1,8 @@ import pytest import json from pathlib import Path +from concurrent.futures import ThreadPoolExecutor +from threading import Event from .conftest import ( api_client_request, @@ -249,6 +251,67 @@ def test_basic_success_case_request(client, tmp_path, api_headers): assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == submitted_url +@pytest.mark.timeout(180) +def test_api_cli_add_concurrent_first_time_default_persona_creation(tmp_path): + """Concurrent live API add requests should share one first-created Default persona.""" + init_archive(tmp_path) + with use_archivebox_db(tmp_path): + from archivebox.personas.models import Persona + + Persona.objects.filter(name="Default").delete() + assert Persona.objects.filter(name="Default").count() == 0 + + port = get_free_port() + env = cli_env(port=port, server=True, USE_COLOR="False", SHOW_PROGRESS="False") + api_token = create_admin_and_token(tmp_path) + submitted_urls = [f"https://example.com/api-cli-add-concurrent-persona-{idx}" for idx in range(4)] + start = Event() + + def post_add(url: str): + start.wait(timeout=10) + return live_api_request( + port, + "post", + "/api/v1/cli/add", + api_token=api_token, + timeout=60, + json={ + "urls": [url], + "depth": 0, + "parser": "url_list", + "plugins": "__archivebox_test_no_plugins__", + "index_only": True, + }, + ) + + try: + start_archivebox_server(tmp_path, env=env, port=port) + with ThreadPoolExecutor(max_workers=len(submitted_urls)) as pool: + futures = [pool.submit(post_add, url) for url in submitted_urls] + start.set() + responses = [future.result(timeout=75) for future in futures] + finally: + stop_server(tmp_path) + + assert [response.status_code for response in responses] == [200] * len(responses), [response.text[:500] for response in responses] + bodies = [response.json() for response in responses] + assert all(body["success"] is True for body in bodies) + assert {body["result"]["queued_urls"][0] for body in bodies} == set(submitted_urls) + + with use_archivebox_db(tmp_path): + from archivebox.personas.models import Persona + + assert Persona.objects.filter(name="Default").count() == 1 + crawls = list(Crawl.objects.order_by("urls").values_list("urls", flat=True)) + root_inputs = sorted( + (snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") + for snapshot in Snapshot.objects.filter(url=Snapshot.INTERNAL_INPUT_URL) + ) + + assert crawls == sorted(submitted_urls) + assert root_inputs == sorted(submitted_urls) + + @pytest.mark.timeout(360) def test_api_cli_add_import_text_formats_preserve_metadata_and_crawl_inner_urls(tmp_path): """REST API add should accept rich import text and queue real inner URLs with metadata preserved.""" diff --git a/archivebox/tests/test_cli_crawl.py b/archivebox/tests/test_cli_crawl.py index a4ba957f..1183c1d6 100644 --- a/archivebox/tests/test_cli_crawl.py +++ b/archivebox/tests/test_cli_crawl.py @@ -417,27 +417,41 @@ def test_crawl_multiple_urls_creates_multiple_snapshots(initialized_archive): assert "https://iana.org" in urls -def test_crawl_from_file_creates_snapshot(initialized_archive): - """Test that crawl can create snapshots from a file of URLs.""" +def test_crawl_path_argument_is_rejected_but_stdin_file_contents_create_snapshot(initialized_archive): + """Local file paths are not URL args; users must pipe file contents through stdin.""" env = cli_env(disable_extractors=True) - # Write URLs to a file urls_file = initialized_archive / "urls.txt" - urls_file.write_text("https://example.com\n") + urls_file.write_text("https://example.com\nhttps://iana.org\n", encoding="utf-8") - run_archivebox_cmd( + path_result = run_archivebox_cmd( ["crawl", "create", str(urls_file)], cwd=initialized_archive, env=env, + ) + assert path_result.returncode == 1 + assert "No URLs provided" in path_result.stderr + + with use_archivebox_db(initialized_archive): + assert Crawl.objects.count() == 0 + assert Snapshot.objects.count() == 0 + + stdin_result = run_archivebox_cmd( + ["crawl", "create"], + stdin=urls_file.read_text(encoding="utf-8"), + cwd=initialized_archive, + env=env, check=True, ) + assert stdin_result.returncode == 0 run_queued_crawls(initialized_archive, env) with use_archivebox_db(initialized_archive): - snapshot = Snapshot.objects.first() + urls = set(Snapshot.objects.values_list("url", flat=True)) - # Should create at least one snapshot (the source file or the URL) - assert snapshot is not None, "Should create at least one snapshot" + assert "https://example.com" in urls + assert "https://iana.org" in urls + assert str(urls_file) not in urls def test_crawl_persists_input_urls_on_crawl(initialized_archive):