Fix runner and persona CI races

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)
 

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()

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:

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,

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."""

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):