Bump release version to 0.9.35rc452

Dockerfile

@@ -10,10 +10,10 @@
 #       --build-context abx-plugins=../abx-plugins \
 #       -t archivebox/abx-dl:dev
 #   docker buildx build . -f Dockerfile \
-#       --build-arg ABX_DL_IMAGE=archivebox/abx-dl:1.12.274 \
+#       --build-arg ABX_DL_IMAGE=archivebox/abx-dl:1.12.276 \
 #       -t archivebox:multistage
 
-ARG ABX_DL_IMAGE=archivebox/abx-dl:1.12.274
+ARG ABX_DL_IMAGE=archivebox/abx-dl:1.12.276
 
 FROM archivebox/sonic:1.4.9 AS sonic
 FROM ${ABX_DL_IMAGE} AS archivebox-runtime-base

archivebox/api/v1_auth.py

@@ -4,7 +4,7 @@ from django.http import HttpRequest
 
 from ninja import Router, Schema
 
-from archivebox.api.auth import auth_using_token, auth_using_password, get_or_create_api_token
+from archivebox.api.auth import auth_using_token, auth_using_password, get_or_create_api_token, HeaderTokenAuth, BearerTokenAuth
 
 
 router = Router(tags=["Authentication"], auth=None)
@@ -63,3 +63,44 @@ def check_api_token(request: HttpRequest, token_data: TokenAuthSchema):
         return {"success": True, "user_id": str(user.pk)}
 
     return {"success": False, "user_id": None}
+
+
+@router.post(
+    "/browser_session",
+    auth=[HeaderTokenAuth(), BearerTokenAuth()],
+    summary="Exchange an administrator API key for a normal browser session",
+)
+def browser_session(request: HttpRequest):
+    from django.conf import settings
+    from django.contrib.auth import login
+    from django.http import JsonResponse
+    from django.utils import timezone
+    from ninja.errors import HttpError
+    from archivebox.core.routes_util import get_admin_base_url
+
+    user = request.auth
+    if not user.is_active or not user.is_superuser:
+        raise HttpError(403, "An active administrator is required")
+    login(request, user, backend="django.contrib.auth.backends.ModelBackend")
+    # A browser session must not outlive the key used to establish it.
+    token = request._api_token
+    lifetime = settings.SESSION_COOKIE_AGE
+    if token.expires:
+        lifetime = min(lifetime, max(0, int((token.expires - timezone.now()).total_seconds())))
+    request.session.set_expiry(lifetime)
+    request.session.save()
+    response = JsonResponse(
+        {
+            "admin_url": get_admin_base_url(request=request).rstrip("/") + "/admin/",
+            "cookie": {
+                "name": settings.SESSION_COOKIE_NAME,
+                "value": request.session.session_key,
+                "expires": request.session.get_expiry_date().timestamp(),
+                "secure": settings.SESSION_COOKIE_SECURE,
+            },
+        },
+    )
+    # The native client installs a host-only HttpOnly cookie on the admin origin;
+    # API and admin hosts can differ, so a Set-Cookie on the API host is insufficient.
+    response["Cache-Control"] = "no-store"
+    return response

archivebox/cli/archivebox_add.py

@@ -388,10 +388,6 @@ def main(**kwargs):
             raise click.BadParameter(str(err), param_hint="--snapshot-max-size") from err
         if kwargs.get("crawl_max_concurrent_snapshots") is not None and int(kwargs["crawl_max_concurrent_snapshots"]) < 1:
             raise click.BadParameter("crawl_max_concurrent_snapshots must be at least 1.", param_hint="--crawl-max-concurrent-snapshots")
-        if not kwargs.get("index_only"):
-            from archivebox.workers.supervisord_util import require_crawl_memory
-
-            require_crawl_memory()
 
         # Translate --only-new/--no-only-new into a crawl config override.
         # add() takes config overrides as a dict; no per-flag kwargs.

archivebox/cli/archivebox_server.py

@@ -363,12 +363,10 @@ def server(
     command = current_command(Process.TypeChoices.SERVER, data_dir=CONSTANTS.DATA_DIR, url=bind_url)
 
     def still_owns_runtime_stack() -> bool:
-        from django.db import connections
-
-        try:
-            return command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR)
-        finally:
-            connections.close_all()
+        # This is polled at the log-tail cadence. Keep the autocommit
+        # connection open so SQLite can reuse its page cache between polls;
+        # reconnecting here repeatedly rereads the process table while idle.
+        return command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR)
 
     shutdown_state = None
     try:
@@ -416,8 +414,13 @@ def server(
     except KeyboardInterrupt:
         pass
     finally:
-        if not shutdown_state or not shutdown_state.signal_name:
-            command.mark_exited()
+        from django.db import connections
+
+        try:
+            if not shutdown_state or not shutdown_state.signal_name:
+                command.mark_exited()
+        finally:
+            connections.close_all()
     print("\n[i][green][🟩] ArchiveBox server shut down gracefully.[/green][/i]")
 
 

archivebox/crawls/admin.py

@@ -1230,7 +1230,7 @@ class CrawlScheduleAdmin(BaseModelAdmin):
         (
             "Configuration",
             {
-                "fields": ("schedule", "template"),
+                "fields": ("is_enabled", "schedule", "template"),
                 "classes": ("card",),
             },
         ),
@@ -1276,7 +1276,21 @@ class CrawlScheduleAdmin(BaseModelAdmin):
 
     def change_view(self, request, object_id, form_url="", extra_context=None):
         self.request = request
-        return super().change_view(request, object_id, form_url, extra_context)
+        from django.db import connections, router
+
+        connection = connections[router.db_for_write(self.model)]
+        if request.method != "POST" or connection.vendor != "sqlite":
+            return super().change_view(request, object_id, form_url, extra_context)
+        # Reserve the writer before Django's atomic admin form reads. A
+        # deferred read transaction cannot safely upgrade while the runner
+        # writes, even when SQLite's existing busy timeout has time left.
+        connection.ensure_connection()
+        previous_mode = connection.transaction_mode
+        connection.transaction_mode = "IMMEDIATE"
+        try:
+            return super().change_view(request, object_id, form_url, extra_context)
+        finally:
+            connection.transaction_mode = previous_mode
 
     def add_view(self, request, form_url="", extra_context=None):
         return redirect("/add/#schedule")

archivebox/search/query.py

@@ -221,6 +221,7 @@ def iter_query_search_ids(
     search_mode: str | None = None,
     config: dict[str, Any] | None = None,
     max_results: int | None = None,
+    stop_event=None,
     **config_kwargs: Any,
 ):
     """Yield snapshot IDs from configured search backend modules."""
@@ -267,6 +268,8 @@ def iter_query_search_ids(
     seen: set[str] = set()
     try:
         for backend_name in backend_names:
+            if stop_event is not None and stop_event.is_set():
+                return
             backend = backends[backend_name]
             try:
                 if backend_name == "sonic":
@@ -292,14 +295,18 @@ def iter_query_search_ids(
                     env=search_backend_command_env(config=config),
                     cwd=CONSTANTS.DATA_DIR,
                     timeout=max(1, int(config.get("TIMEOUT", 60))) * 4,
+                    stop_event=stop_event,
                 )
-                for snapshot_id in ids:
-                    if snapshot_id in seen:
-                        continue
-                    seen.add(snapshot_id)
-                    yield snapshot_id
-                    if max_results and len(seen) >= max_results:
-                        return
+                try:
+                    for snapshot_id in ids:
+                        if snapshot_id in seen:
+                            continue
+                        seen.add(snapshot_id)
+                        yield snapshot_id
+                        if max_results and len(seen) >= max_results:
+                            return
+                finally:
+                    ids.close()
                 successful_backends += 1
             except Exception as err:
                 errors.append(err)

archivebox/search/views.py

@@ -6,7 +6,7 @@ import json
 import threading
 import time
 from copy import copy
-from queue import Full, Queue
+from queue import Empty, Full, Queue
 from urllib.parse import urlsplit
 from uuid import UUID
 
@@ -254,15 +254,18 @@ def iter_filtered_search_result_ids(iterator, queryset, *, flush_max_delay=0.05)
         yield from flush_batch()
 
 
-def iter_search_result_ids(query, base_queryset, *, search_mode, config):
+def iter_search_result_ids(query, base_queryset, *, search_mode, config, stop_event=None):
     """Yield filtered Snapshot IDs from the selected search provider."""
     search_mode_base = get_search_mode_base(search_mode, config=config)
     provider = (
         iter_meta_search_ids(query, base_queryset)
         if search_mode_base == "meta"
-        else iter_query_search_ids(query, search_mode=search_mode, config=config)
+        else iter_query_search_ids(query, search_mode=search_mode, config=config, stop_event=stop_event)
     )
-    yield from iter_filtered_search_result_ids(provider, base_queryset)
+    try:
+        yield from iter_filtered_search_result_ids(provider, base_queryset)
+    finally:
+        provider.close()
 
 
 def snapshot_search_stream_response(query, base_queryset, *, search_mode, config, cache_key, thread_name):
@@ -300,7 +303,7 @@ def snapshot_search_stream_response(query, base_queryset, *, search_mode, config
             iterator = None
             try:
                 close_old_connections()
-                iterator = iter_search_result_ids(query, base_queryset, search_mode=search_mode, config=config)
+                iterator = iter_search_result_ids(query, base_queryset, search_mode=search_mode, config=config, stop_event=stop_event)
                 for snapshot_id in iterator:
                     if stop_event.is_set():
                         break
@@ -326,10 +329,13 @@ def snapshot_search_stream_response(query, base_queryset, *, search_mode, config
                 emit(None)
 
         threading.Thread(target=run_search, name=thread_name, daemon=True).start()
-        yield f"0{stream_padding}\n"
         try:
+            yield f"0{stream_padding}\n"
             while True:
-                item = await asyncio.to_thread(queue.get)
+                try:
+                    item = await asyncio.to_thread(queue.get, True, 0.1)
+                except Empty:
+                    continue
                 if item is None:
                     break
                 if isinstance(item, BaseException):

archivebox/services/resource_admission.py

@@ -0,0 +1,91 @@
+"""Admission signals, not estimates of how much RAM an extractor will need."""
+
+import math
+import os
+import time
+from pathlib import Path
+
+
+RESOURCE_RECHECK_SECONDS = 2.0
+# Scheduling policy, not a minimum-RAM claim: stop increasing extraction
+# load when all non-idle work loses at least one second out of ten to memory
+# stalls. Require both recent and rolling pressure so transient reclaim does
+# not indefinitely starve otherwise viable work on small hosts.
+MEMORY_FULL_STALL_PERCENT = 10.0
+
+
+class ResourceAdmission:
+    def __init__(self):
+        self._memory_stalls: dict[str, tuple[int, float]] = {}
+        self._checked_at = float("-inf")
+        self._reason: str | None = None
+
+    @staticmethod
+    def cgroup_paths() -> list[Path]:
+        root = Path("/sys/fs/cgroup")
+        paths = [root]
+        try:
+            for line in Path("/proc/self/cgroup").read_text().splitlines():
+                if line.startswith("0::"):
+                    current = root / line[3:].lstrip("/")
+                    if current.is_dir() and current.is_relative_to(root):
+                        paths = [current, *[parent for parent in current.parents if parent.is_relative_to(root)]]
+                    break
+        except OSError:
+            pass
+        return paths
+
+    def snapshot_limit(self, configured: int) -> int:
+        cpus = len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else (os.cpu_count() or 1)
+        for path in self.cgroup_paths():
+            try:
+                quota, period = (path / "cpu.max").read_text().split()
+                if quota != "max":
+                    cpus = min(cpus, max(1, math.ceil(int(quota) / int(period))))
+            except (OSError, ValueError, ZeroDivisionError):
+                continue
+        # CPU quota limits parallel admission, not forward progress. A host
+        # with a fractional CPU must still be able to run one snapshot.
+        return max(1, min(configured, cpus))
+
+    def observe_memory_stalls(self, source: str, total: int, avg10: float, sampled_at: float) -> bool:
+        previous = self._memory_stalls.get(source)
+        self._memory_stalls[source] = (total, sampled_at)
+        if previous is None:
+            return avg10 >= MEMORY_FULL_STALL_PERCENT
+        previous_total, previous_time = previous
+        elapsed = sampled_at - previous_time
+        if elapsed <= 0 or total < previous_total:
+            return False
+        stalled_percent = (total - previous_total) / (elapsed * 1_000_000) * 100
+        return avg10 >= MEMORY_FULL_STALL_PERCENT and stalled_percent >= MEMORY_FULL_STALL_PERCENT
+
+    def memory_pressure(self) -> str | None:
+        now = time.monotonic()
+        if now - self._checked_at < RESOURCE_RECHECK_SECONDS:
+            return self._reason
+        self._checked_at = now
+        reasons = []
+        cgroups = self.cgroup_paths()
+        for path in cgroups:
+            try:
+                current = int((path / "memory.current").read_text())
+                for name in ("memory.high", "memory.max"):
+                    limit = (path / name).read_text().strip()
+                    if limit != "max" and current >= int(limit):
+                        reasons.append(f"{name} reached")
+            except (OSError, ValueError):
+                continue
+        for path in [Path("/proc/pressure/memory"), *[p / "memory.pressure" for p in cgroups]]:
+            try:
+                full = next(line for line in path.read_text().splitlines() if line.startswith("full "))
+                fields = dict(field.split("=", 1) for field in full.split()[1:])
+                if self.observe_memory_stalls(str(path), int(fields["total"]), float(fields["avg10"]), now):
+                    reasons.append("memory stalls")
+            except (OSError, ValueError, KeyError, StopIteration):
+                continue
+        self._reason = ", ".join(dict.fromkeys(reasons)) or None
+        return self._reason
+
+
+resource_admission = ResourceAdmission()

archivebox/services/runner.py

@@ -58,6 +58,8 @@ from abxbus import BaseEvent
 from abxbus.event_bus import EventBus, get_current_event, in_handler_context
 from abxbus.event_handler import EventHandlerAbortedError, EventHandlerCancelledError
 
+from archivebox.services.resource_admission import RESOURCE_RECHECK_SECONDS, resource_admission
+
 from archivebox.config.common import (
     ArchiveBoxBaseConfig,
     normalize_runtime_config,
@@ -221,6 +223,8 @@ class CrawlRunner:
         self.snapshot_tasks: dict[str, asyncio.Task[None]] = {}
         self.snapshot_semaphore = asyncio.Semaphore(1)
         self.max_concurrent_snapshots = 1
+        self._resource_deferred = False
+        self._initial_snapshot_admitted = False
         self.persona = None
         self.base_config: ArchiveBoxBaseConfig | dict[str, Any] = {}
         self.derived_config: dict[str, Any] = {}
@@ -294,6 +298,7 @@ class CrawlRunner:
     async def run(self) -> None:
         root_snapshot_id: str | None = None
         bus_destroyed = False
+        run_state_loaded = False
         try:
             first_signal_message = (
                 "\n[🛑] Got {signal_name}, aborting the active hook...\n"
@@ -313,8 +318,15 @@ class CrawlRunner:
                 on_signal=self._request_abort_from_signal,
                 raise_on_first_signal=False,
             ):
+                if await sync_to_async(defer_crawl_for_resources, thread_sensitive=True)(self.crawl):
+                    return
+                # Startup and its first snapshot are one admission. Browser
+                # startup can itself cause reclaim; checking again immediately
+                # would tear it down and repeat startup without doing any work.
+                self._initial_snapshot_admitted = True
+                run_state_loaded = True
                 snapshot_ids = await sync_to_async(self.load_run_state, thread_sensitive=True)()
-                max_concurrent_snapshots = max(1, int(self.base_config.get("CRAWL_MAX_CONCURRENT_SNAPSHOTS", 1)))
+                max_concurrent_snapshots = resource_admission.snapshot_limit(int(self.base_config.get("CRAWL_MAX_CONCURRENT_SNAPSHOTS", 1)))
                 self.max_concurrent_snapshots = max_concurrent_snapshots
                 self.snapshot_semaphore = asyncio.Semaphore(max_concurrent_snapshots)
                 live_ui = self._create_live_ui()
@@ -344,10 +356,19 @@ class CrawlRunner:
                 except Exception:
                     pass
                 self._live_stream = None
-            await sync_to_async(project_abxpkg_derived_cache_to_db, thread_sensitive=True)(self.base_config.get("ABXPKG_LIB_DIR"))
-            await sync_to_async(self.finalize_run_state, thread_sensitive=True)()
+            if run_state_loaded:
+                await sync_to_async(project_abxpkg_derived_cache_to_db, thread_sensitive=True)(self.base_config.get("ABXPKG_LIB_DIR"))
+                await sync_to_async(self.finalize_run_state, thread_sensitive=True)()
 
     async def enqueue_snapshot(self, snapshot_id: str, crawl_start_event: CrawlStartEvent | None = None) -> None:
+        if self._resource_deferred:
+            return
+        if not self._initial_snapshot_admitted and await sync_to_async(defer_crawl_for_resources, thread_sensitive=True)(
+            self.crawl,
+            excluded_snapshot_ids=list(self.snapshot_tasks),
+        ):
+            self._resource_deferred = True
+            return
         if await self.crawl_is_cancelled():
             return
         if await self.crawl_is_paused() and not self.allow_maintenance_on_inactive_crawl:
@@ -362,6 +383,7 @@ class CrawlRunner:
             return
         else:
             task = asyncio.create_task(self.run_snapshot(snapshot_id), context=_runner_task_context())
+        self._initial_snapshot_admitted = False
         self.snapshot_tasks[snapshot_id] = task
 
     async def stop_snapshot_tasks(self) -> None:
@@ -488,6 +510,8 @@ class CrawlRunner:
         from archivebox.core.models import Snapshot
         from archivebox.config.common import get_config
 
+        if self._resource_deferred:
+            return
         if not isinstance(get_current_event(), CrawlStartEvent):
             return
         if await self.crawl_is_cancelled():
@@ -497,7 +521,7 @@ class CrawlRunner:
 
         await sync_to_async(self.crawl.refresh_from_db, thread_sensitive=True)()
         config = await sync_to_async(lambda: get_config(crawl=self.crawl), thread_sensitive=True)()
-        self.max_concurrent_snapshots = max(1, int(config["CRAWL_MAX_CONCURRENT_SNAPSHOTS"]))
+        self.max_concurrent_snapshots = resource_admission.snapshot_limit(int(config["CRAWL_MAX_CONCURRENT_SNAPSHOTS"]))
 
         active_snapshot_ids = [snapshot_id for snapshot_id, task in self.snapshot_tasks.items() if not task.done()]
         available_slots = max(0, self.max_concurrent_snapshots - len(active_snapshot_ids))
@@ -1266,6 +1290,29 @@ def run_snapshot_maintenance(snapshot_id: str, *, output_dir: Path | None = None
     return True
 
 
+def defer_crawl_for_resources(crawl, *, excluded_snapshot_ids=()) -> bool:
+    """Leave extraction durable; never delay cancellation or final cleanup."""
+    from archivebox.core.models import Snapshot
+
+    if crawl.status not in (crawl.StatusChoices.QUEUED, crawl.StatusChoices.STARTED):
+        return False
+    reason = resource_admission.memory_pressure()
+    if reason is None:
+        return False
+    pending = crawl.snapshot_set.filter(status__in=Snapshot.RUNNABLE_STATES).exclude(id__in=excluded_snapshot_ids)
+    if crawl.snapshot_set.exists() and not pending.exists():
+        return False
+    now = timezone.now()
+    retry_at = now + timedelta(seconds=RESOURCE_RECHECK_SECONDS)
+    # Conditional short updates cannot resurrect a concurrent pause/cancel.
+    type(crawl).objects.filter(id=crawl.id, status__in=[crawl.StatusChoices.QUEUED, crawl.StatusChoices.STARTED]).update(
+        retry_at=retry_at,
+        modified_at=now,
+    )
+    pending.filter(retry_at__lte=now).update(retry_at=retry_at, modified_at=now)
+    return True
+
+
 def run_due_crawl(crawl, *, lock_seconds: int, interactive_interrupts: bool = False) -> bool:
     with crawl_lifecycle_lock(str(crawl.id)):
         return _run_due_crawl_locked(

archivebox/templates/core/system_warnings_banner.html

@@ -34,7 +34,7 @@
     {% elif mode == "high_load" %}
       15-min loadavg <code style="background:#000;color:#fff;padding:1px 6px;border-radius:3px;font:700 11px/1.3 ui-monospace,Menlo,Consolas,monospace;text-transform:none;">{{ load_15 }}</code>
       exceeds <code style="background:#000;color:#fff;padding:1px 6px;border-radius:3px;font:700 11px/1.3 ui-monospace,Menlo,Consolas,monospace;text-transform:none;">{{ load_threshold }}</code>
-      ({{ cpu_count }} cores × 3) — the host is saturated; crawls are queuing faster than the runner can finish them.
+      ({{ cpu_count }} cores × 3) — recent CPU or I/O contention may slow crawls. This average can stay elevated after load subsides; it does not measure queue growth.
     {% elif mode == "unsafe" %}
       ArchiveBox single-domain mode — archived pages share an origin with this site
       <a href="https://github.com/ArchiveBox/ArchiveBox/wiki/Security-Modes" target="_blank" rel="noopener noreferrer" style="color:#fff;text-decoration:underline;margin-left:6px;">why?</a>

archivebox/tests/test_api_v1_auth_browser_session.py

@@ -0,0 +1,56 @@
+from datetime import timedelta
+from urllib.parse import urlsplit
+
+import pytest
+from django.contrib.sessions.models import Session
+from django.utils import timezone
+
+from archivebox.api.models import APIToken
+from archivebox.tests.conftest import API_TEST_HOST, api_client_request
+
+pytestmark = pytest.mark.django_db(transaction=True)
+
+
+@pytest.mark.parametrize("header", ["HTTP_AUTHORIZATION", "HTTP_X_ARCHIVEBOX_API_KEY"])
+def test_browser_session_authenticates_admin_and_expires_with_key(client, api_admin_user, header):
+    token = APIToken.objects.create(created_by=api_admin_user, expires=timezone.now() + timedelta(minutes=5))
+    response = api_client_request(
+        client,
+        "post",
+        "/api/v1/auth/browser_session",
+        payload={},
+        headers={
+            "HTTP_HOST": API_TEST_HOST,
+            header: f"Bearer {token.token}" if header == "HTTP_AUTHORIZATION" else token.token,
+        },
+    )
+    assert response.status_code == 200, response.content
+    assert response["Cache-Control"] == "no-store"
+    data = response.json()
+    session = Session.objects.get(session_key=data["cookie"]["value"])
+    assert session.get_decoded()["_auth_user_id"] == str(api_admin_user.pk)
+    assert timezone.now() < session.expire_date <= token.expires
+    client.cookies.clear()
+    client.cookies[data["cookie"]["name"]] = data["cookie"]["value"]
+    response = client.get("/admin/", HTTP_HOST=urlsplit(data["admin_url"]).netloc)
+    assert response.status_code == 200, response.content
+    assert response.wsgi_request.user.pk == api_admin_user.pk
+
+
+@pytest.mark.parametrize("kind", ["missing", "invalid", "expired", "inactive", "nonadmin", "query"])
+def test_browser_session_rejects_invalid_credentials(client, api_admin_user, kind):
+    token = APIToken.objects.create(created_by=api_admin_user, expires=timezone.now() + timedelta(minutes=5))
+    if kind == "expired":
+        token.expires = timezone.now() - timedelta(seconds=1)
+        token.save()
+    if kind in ("inactive", "nonadmin"):
+        setattr(api_admin_user, "is_active" if kind == "inactive" else "is_superuser", False)
+        api_admin_user.save()
+    headers = {"HTTP_HOST": API_TEST_HOST}
+    if kind not in ("missing", "query"):
+        headers["HTTP_AUTHORIZATION"] = "Bearer " + ("invalid" if kind == "invalid" else token.token)
+    path = "/api/v1/auth/browser_session" + (f"?api_key={token.token}" if kind == "query" else "")
+    before = Session.objects.count()
+    response = api_client_request(client, "post", path, payload={}, headers=headers)
+    assert response.status_code in (401, 403), response.content
+    assert Session.objects.count() == before

archivebox/tests/test_cli_add.py

@@ -230,27 +230,23 @@ def test_add_single_url_records_url_in_crawl(initialized_archive):
     assert snapshots == []
 
 
-def test_add_memory_preflight_fails_before_creating_crawl(initialized_archive, capsys):
-    from archivebox.workers.supervisord_util import MIN_CRAWL_AVAILABLE_MEMORY_BYTES, require_crawl_memory
-
-    assert MIN_CRAWL_AVAILABLE_MEMORY_BYTES == 512 * 1024 * 1024
-    require_crawl_memory(MIN_CRAWL_AVAILABLE_MEMORY_BYTES)
-
+def test_add_bg_persists_queue_without_starting_extraction_workers(initialized_archive):
+    result = run_archivebox_cmd(
+        ["add", "--bg", "--plugins=wget", "--crawl-max-concurrent-snapshots=4", "https://example.com", "https://example.org"],
+        cwd=initialized_archive,
+        env=cli_env(),
+    )
+    assert result.returncode == 0, result.stderr
     with use_archivebox_db(initialized_archive):
-        before_crawls = Crawl.objects.count()
-        before_processes = Process.objects.count()
-
-        with pytest.raises(SystemExit) as err:
-            require_crawl_memory(MIN_CRAWL_AVAILABLE_MEMORY_BYTES - 1)
-
-        assert Crawl.objects.count() == before_crawls
-        assert Process.objects.count() == before_processes
-
-    assert err.value.code == 1
-    output = capsys.readouterr().err
-    assert "Not enough available memory to archive a crawl" in output
-    assert "No crawl, runner, or Sonic workers were started" in output
-    assert "configure swap" in output
+        crawl = Crawl.objects.get()
+        assert crawl.status == Crawl.StatusChoices.QUEUED
+        assert crawl.retry_at is not None
+        assert crawl.get_urls_list() == ["https://example.com", "https://example.org"]
+        assert crawl.config["PLUGINS"] == "wget"
+        assert crawl.config["CRAWL_MAX_CONCURRENT_SNAPSHOTS"] == 4
+        assert not Snapshot.objects.exists()
+        assert not ArchiveResult.objects.exists()
+        assert not Process.objects.filter(process_type__in=[Process.TypeChoices.ORCHESTRATOR, Process.TypeChoices.HOOK]).exists()
 
 
 def test_initial_crawl_creation_does_not_warn_as_an_outside_runner_update(initialized_archive, caplog):

archivebox/tests/test_resource_admission.py

@@ -0,0 +1,30 @@
+import os
+
+from archivebox.services.resource_admission import ResourceAdmission
+
+
+def test_cpu_admission_respects_process_affinity():
+    admission = ResourceAdmission()
+    available = len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else os.cpu_count()
+    assert 1 <= admission.snapshot_limit(100000) <= available
+    assert admission.snapshot_limit(1) == 1
+
+
+def test_memory_stall_counter_recovers_without_resetting_monitor():
+    # Real kernel PSI units: cumulative microseconds and ten-second average.
+    # Decision samples exercise the policy, not simulated hooks/processes.
+    admission = ResourceAdmission()
+    assert admission.observe_memory_stalls("host", 100, 0.0, 0.0) is False
+    assert admission.observe_memory_stalls("host", 400100, 20.0, 2.0) is True
+    assert admission.observe_memory_stalls("host", 400100, 20.0, 4.0) is False
+    assert admission.observe_memory_stalls("new-cgroup", 900, 20.0, 0.0) is True
+    assert admission.observe_memory_stalls("new-cgroup", 900, 20.0, 2.0) is False
+
+
+def test_transient_and_low_memory_stalls_do_not_starve_admission():
+    admission = ResourceAdmission()
+    for sample in range(20):
+        assert admission.observe_memory_stalls("host", sample * 2600, 0.13, sample * 2.0) is False
+    # One large interval alone is not sustained pressure; the rolling average
+    # must agree before deferring work.
+    assert admission.observe_memory_stalls("host", 1_000_000, 0.13, 40.0) is False

archivebox/tests/test_search.py

@@ -1289,3 +1289,75 @@ class TestSearchBackendsE2E:
             fixture_server.shutdown()
             fixture_server.server_close()
             fixture_thread.join(timeout=5)
+
+
+@pytest.mark.timeout(90)
+def test_live_public_search_disconnect_reaps_backend(initialized_archive):
+    """A real HTTP disconnect cancels a silent ripgrep process and its wrapper."""
+    import signal
+    from uuid import uuid4
+
+    import psutil
+
+    from archivebox.core.models import Snapshot
+    from archivebox.tests.test_orm_helpers import use_archivebox_db
+
+    rg = shutil.which("rg")
+    assert rg
+    query = "disconnect" + uuid4().hex
+    port = get_free_port()
+    env = cli_env(
+        live=True,
+        PUBLIC_INDEX="True",
+        PERMISSIONS="public",
+        SEARCH_BACKEND_ENGINE="ripgrep",
+        SEARCH_BACKEND_RIPGREP_ENABLED="True",
+        RIPGREP_BINARY=rg,
+        RIPGREP_ARGS_EXTRA='["--no-mmap", "--text", "--sort", "path"]',
+        RIPGREP_TIMEOUT="30",
+        ALLOWED_HOSTS="*",
+    )
+    created = run_archivebox_cmd(
+        ["snapshot", "create", "--status=sealed", "https://example.com/disconnect-test"],
+        cwd=initialized_archive,
+        env=env,
+    )
+    assert created.returncode == 0, created.stderr
+    with use_archivebox_db(initialized_archive):
+        snapshot = Snapshot.objects.get(url="https://example.com/disconnect-test")
+        directory = snapshot.output_dir / "dom"
+        directory.mkdir(parents=True, exist_ok=True)
+        (directory / "a.txt").write_text(query + "\n")
+        with (directory / "z.txt").open("wb") as large:
+            large.truncate(8 * 1024**3)
+            for offset in range(0, 8 * 1024**3, 1024**2):
+                large.seek(offset)
+                large.write(b"\n")
+    server = start_archivebox_server(initialized_archive, port=port, env=env, daemonize=False, log_name="disconnect-server.log")
+    children = []
+    try:
+        with requests.get(
+            f"http://127.0.0.1:{port}/public/search-stream/",
+            params={"q": query, "search_mode": "deep:ripgrep"},
+            headers={"Host": f"web.archivebox.localhost:{port}"},
+            stream=True,
+            timeout=(5, 10),
+        ) as response:
+            assert response.status_code == 200, response.text
+            lines = response.iter_lines()
+            assert int(next(lines).strip()) == 0
+            assert int(next(lines).strip()) == 1
+            children = [child for child in psutil.Process(server.pid).children(recursive=True) if query in " ".join(child.cmdline())]
+            assert len(children) == 2
+            for child in children:
+                if Path(child.exe()).name == Path(rg).name:
+                    child.send_signal(signal.SIGSTOP)
+        _, alive = psutil.wait_procs(children, timeout=5)
+        assert not alive, [(child.pid, child.status()) for child in alive]
+    finally:
+        for child in children:
+            try:
+                child.kill()
+            except psutil.NoSuchProcess:
+                pass
+        stop_archivebox_process(server)

archivebox/tests/test_ui_admin_crawl.py

@@ -14,6 +14,7 @@ pytestmark = pytest.mark.django_db
 
 
 class TestCrawlScheduleAdmin:
+    @pytest.mark.django_db(transaction=True)
     def test_crawlschedule_change_view_renders_and_saves(self, client, admin_user, crawl):
         from archivebox.crawls.models import CrawlSchedule
 
@@ -35,24 +36,33 @@ class TestCrawlScheduleAdmin:
         assert b"No Crawls yet..." not in get_response.content
         assert b"No Snapshots yet..." not in get_response.content
 
-        post_response = client.post(
-            change_url,
-            {
-                "label": "Morning crawl",
-                "notes": "updated",
-                "schedule": "0 8 * * *",
-                "template": str(crawl.pk),
-                "created_by": str(admin_user.pk),
-                "_save": "Save",
-            },
-            HTTP_HOST=ADMIN_TEST_HOST,
-        )
+        from django.db import connection
+        from django.test.utils import CaptureQueriesContext
+
+        previous_mode = getattr(connection, "transaction_mode", None)
+        with CaptureQueriesContext(connection) as queries:
+            post_response = client.post(
+                change_url,
+                {
+                    "label": "Morning crawl",
+                    "notes": "updated",
+                    "schedule": "0 8 * * *",
+                    "template": str(crawl.pk),
+                    "created_by": str(admin_user.pk),
+                    "_save": "Save",
+                },
+                HTTP_HOST=ADMIN_TEST_HOST,
+            )
+        if connection.vendor == "sqlite":
+            assert any(query["sql"] == "BEGIN IMMEDIATE" for query in queries)
+            assert connection.transaction_mode == previous_mode
 
         assert post_response.status_code == 302
         schedule.refresh_from_db()
         assert schedule.label == "Morning crawl"
         assert schedule.notes == "updated"
         assert schedule.schedule == "0 8 * * *"
+        assert schedule.is_enabled is False
         assert schedule.template_id == crawl.pk
         assert schedule.created_by_id == admin_user.pk
 

archivebox/workers/supervisord_util.py

@@ -47,7 +47,6 @@ _SUPERVISORD_ERRORS = (XmlRpcError, OSError, RuntimeError, TimeoutError)
 _PROCESS_STATE_ERRORS = (DatabaseError, OSError, RuntimeError, ValueError, psutil.Error)
 MIN_SERVER_WORKER_AVAILABLE_MEMORY_BYTES = 256 * 1024 * 1024
 MIN_DEPENDENCY_INSTALL_AVAILABLE_MEMORY_BYTES = MIN_SERVER_WORKER_AVAILABLE_MEMORY_BYTES
-MIN_CRAWL_AVAILABLE_MEMORY_BYTES = 512 * 1024 * 1024
 
 
 def _shell_join(args: list[str]) -> str:
@@ -118,16 +117,6 @@ def require_dependency_install_memory(available_bytes: int | None = None) -> Non
     )
 
 
-def require_crawl_memory(available_bytes: int | None = None) -> None:
-    available_bytes = effective_available_memory_bytes() if available_bytes is None else available_bytes
-    _require_available_memory(
-        "archive a crawl",
-        "    No crawl, runner, or Sonic workers were started.",
-        available_bytes,
-        MIN_CRAWL_AVAILABLE_MEMORY_BYTES,
-    )
-
-
 def archivebox_cmd(*args: str) -> list[str]:
     return [str(resolve_env_binary("archivebox")), *args]
 
@@ -1280,11 +1269,18 @@ def tail_multiple_worker_logs(log_files: list[str], follow=True, proc=None, keep
 
     print()
 
+    # Log display needs a short refresh interval, but each ownership check
+    # queries the database and validates OS processes. Check leadership once
+    # per second even when workers continuously produce log output.
+    next_ownership_check = 0.0
     try:
         while follow:
-            if keep_running is not None and not keep_running():
-                print("\n[newer ArchiveBox process is now running the orchestrator and server]")
-                return "transferred"
+            now = time.monotonic()
+            if keep_running is not None and now >= next_ownership_check:
+                if not keep_running():
+                    print("\n[newer ArchiveBox process is now running the orchestrator and server]")
+                    return "transferred"
+                next_ownership_check = time.monotonic() + 1.0
 
             # Check if the monitored process has exited
             if proc is not None and proc.poll() is not None:

etc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "archivebox",
-  "version": "0.9.35rc449",
+  "version": "0.9.35rc452",
   "repository": "github:ArchiveBox/ArchiveBox",
   "license": "MIT",
   "dependencies": {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "archivebox"
-version = "0.9.35rc449"
+version = "0.9.35rc452"
 requires-python = ">=3.13"
 description = "Self-hosted internet archiving solution."
 authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}]
@@ -79,8 +79,8 @@ dependencies = [
     ### Binary/Package Management
     "abxbus==2.5.56",                 # direct imports only; version constrained by abx-dl -> abx-plugins
     "abxpkg==1.12.117",                # direct imports only; version constrained by abx-dl -> abx-plugins
-    "abx-plugins[opencode]==1.12.236", # optional AI runtime clients are owned by the plugin
-    "abx-dl==1.12.274",       # shared ArchiveBox downloader package
+    "abx-plugins[opencode]==1.12.237", # optional AI runtime clients are owned by the plugin
+    "abx-dl==1.12.276",       # shared ArchiveBox downloader package
     ### UUID7 backport for Python <3.14
     "uuid7>=0.1.0; python_version < '3.14'",  # provides the uuid_extensions module on Python 3.13
     "browser-cookie3>=0.20.1",
@@ -342,7 +342,7 @@ Donate = "https://github.com/ArchiveBox/ArchiveBox/wiki/Donations"
 
 
 [tool.bumpver]
-current_version = "v0.9.35rc449"
+current_version = "v0.9.35rc452"
 version_pattern = "vMAJOR.MINOR.PATCH[PYTAGNUM]"
 commit_message = "bump version {old_version} -> {new_version}"
 tag_message = "{new_version}"

uv.lock

@@ -13,18 +13,18 @@ supported-markers = [
 ]
 
 [options]
-exclude-newer = "2026-09-11T22:03:43.68620639Z"
+exclude-newer = "2026-09-12T20:05:07.519727728Z"
 exclude-newer-span = "P5D"
 
 [options.exclude-newer-package]
-abxbus = { timestamp = "2026-09-16T22:03:42.686219334Z", span = "PT1S" }
+abxbus = { timestamp = "2026-09-17T20:05:06.519740057Z", span = "PT1S" }
 abx-plugins = "2100-01-01T00:00:00Z"
 abx-dl = "2100-01-01T00:00:00Z"
 abxpkg = "2100-01-01T00:00:00Z"
 
 [[package]]
 name = "abx-dl"
-version = "1.12.274"
+version = "1.12.276"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "abx-plugins", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
@@ -38,14 +38,14 @@ dependencies = [
     { name = "rich", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
     { name = "rich-click", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/c6/69/223fb0cfd106eda015a1a73daa58fc9aa87bb7b974094c5eadc65c15b693/abx_dl-1.12.274.tar.gz", hash = "sha256:58b7f85aec60db8233d4ea33a7683edf12be63fe32b8eccbd45d57edcab1a35b", size = 87257 }
+sdist = { url = "https://files.pythonhosted.org/packages/2a/0d/cd3de0fabd859f19775a6bcd4e4bfae94f855243401826f8efc2749e330a/abx_dl-1.12.276.tar.gz", hash = "sha256:1248f14a24f9f592f890ee79aec6838116095dec2388941439c000f0f20b94a9", size = 88238 }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ab/36/b12628d4ce600cbcd1c3ba87c6f76721a1c21e074e1f30782baf2eb2eb06/abx_dl-1.12.274-py3-none-any.whl", hash = "sha256:5481d25ab82da2449a3a0a47079fd37a8768f1175ed2068bef45392beb4ee684", size = 92246 },
+    { url = "https://files.pythonhosted.org/packages/ae/bb/e5ba379e6626d67c76ce1608303289b4460a42147d5e81c2769da1b8f2a2/abx_dl-1.12.276-py3-none-any.whl", hash = "sha256:876105e13945d9713e62c8aad220c5a33ebb918c482af28e3662a25e9a86859a", size = 93263 },
 ]
 
 [[package]]
 name = "abx-plugins"
-version = "1.12.236"
+version = "1.12.237"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "abxbus", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
@@ -54,9 +54,9 @@ dependencies = [
     { name = "rich-click", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
     { name = "uv", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/11/90/617fe56f24af0ed1a6bcc255233703c9378f49d381d2192eb76a0a1c9ad7/abx_plugins-1.12.236.tar.gz", hash = "sha256:1c0a23a4878787a64109c69160c71df74885c90e5fca9def7c13b8ea731b1add", size = 269546, upload-time = "2026-09-16T21:51:20.39Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/dd/f2/9a2deb1a777ee1230cfd4e41345b69bcb0a271ec3b78e12c3ccf2c0e9066/abx_plugins-1.12.237.tar.gz", hash = "sha256:2eb0734bba0b797c95e1f370bbb833b7b51c101ac98809fb98c9db905f1c5c95", size = 270072, upload-time = "2026-09-17T19:43:40.469Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/6c/ed/ac9682624eea2604cb3d35ac6f952c92c95e52c064401044d43c3df5864a/abx_plugins-1.12.236-py3-none-any.whl", hash = "sha256:0bf8738c64d27dd88d2ac4e91c4c85504352cc1a47bdc9aa863a12a9b7f8a5da", size = 424591, upload-time = "2026-09-16T21:51:21.816Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/8c/23030e3483c6100e418174227f874a0a1fd8a5ad4772e8267316e8ba4193/abx_plugins-1.12.237-py3-none-any.whl", hash = "sha256:b59923d87228d0fc1f50f52867cf27c4508fd6c6a114fd7c44f986fe46d4ac96", size = 425064, upload-time = "2026-09-17T19:43:42.081Z" },
 ]
 
 [package.optional-dependencies]
@@ -127,7 +127,7 @@ wheels = [
 
 [[package]]
 name = "archivebox"
-version = "0.9.35rc449"
+version = "0.9.35rc452"
 source = { editable = "." }
 dependencies = [
     { name = "abx-dl", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" },
@@ -226,8 +226,8 @@ dev = [
 
 [package.metadata]
 requires-dist = [
-    { name = "abx-dl", specifier = "==1.12.274" },
-    { name = "abx-plugins", extras = ["opencode"], specifier = "==1.12.236" },
+    { name = "abx-dl", specifier = "==1.12.276" },
+    { name = "abx-plugins", extras = ["opencode"], specifier = "==1.12.237" },
     { name = "abxbus", specifier = "==2.5.56" },
     { name = "abxpkg", specifier = "==1.12.117" },
     { name = "archivebox", extras = ["sonic", "ldap", "debug"], marker = "extra == 'all'" },