From 75441eb8d71821bd2cf19a908959ab70baca4108 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Mon, 1 Jun 2026 10:37:44 -0700 Subject: [PATCH] fix: restart supervised runner from add view --- archivebox/services/runner.py | 10 +- archivebox/tests/test_ui_add_view_runtime.py | 128 +++++++++++++++++++ 2 files changed, 137 insertions(+), 1 deletion(-) diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 6d3bdeb0..c4a8e8b2 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -181,7 +181,7 @@ def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool: from archivebox.config import CONSTANTS from archivebox.machine.models import Machine, Process - from archivebox.workers.supervisord_util import get_existing_supervisord_process, get_worker + from archivebox.workers.supervisord_util import RUNNER_WORKER, get_existing_supervisord_process, get_worker, start_worker supervisor = get_existing_supervisord_process() runner_worker = get_worker(supervisor, "worker_runner") if supervisor else None @@ -189,6 +189,7 @@ def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool: return False machine = Machine.current() + Process.cleanup_stale_running(machine=machine) running_orchestrators = Process.objects.filter( machine=machine, status=Process.StatusChoices.RUNNING, @@ -197,6 +198,13 @@ def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool: if any(proc.is_running for proc in running_orchestrators): return False + if supervisor is not None: + try: + start_worker(supervisor, RUNNER_WORKER) + return True + except Exception: + pass + log_path = CONSTANTS.LOGS_DIR / "errors.log" log_path.parent.mkdir(parents=True, exist_ok=True) env = os.environ.copy() diff --git a/archivebox/tests/test_ui_add_view_runtime.py b/archivebox/tests/test_ui_add_view_runtime.py index 6c46d501..1bec66a8 100644 --- a/archivebox/tests/test_ui_add_view_runtime.py +++ b/archivebox/tests/test_ui_add_view_runtime.py @@ -14,6 +14,7 @@ from .conftest import ( get_depth_counts, get_free_port, init_archive, + run_archivebox_cmd_cwd, start_server, stop_server, wait_for_http, @@ -22,6 +23,133 @@ from .conftest import ( pytestmark = pytest.mark.django_db(transaction=True) +@pytest.mark.timeout(180) +def test_add_view_restarts_stopped_supervisord_runner(tmp_path, recursive_test_site): + os.chdir(tmp_path) + init_archive(tmp_path) + + port = get_free_port() + env = build_test_env( + port, + PLUGINS="wget", + PUBLIC_ADD_VIEW="True", + PYTEST_CURRENT_TEST="", + ) + create_admin_and_token(tmp_path) + + try: + start_server(tmp_path, env=env, port=port) + _wait_for_worker_state(tmp_path, "worker_runner", "RUNNING") + _stop_worker(tmp_path, "worker_runner") + assert _worker_state(tmp_path, "worker_runner") != "RUNNING" + + session, csrf_token = _login_to_add_view(port) + response = session.post( + f"http://127.0.0.1:{port}/add/", + headers={"Host": f"admin.archivebox.localhost:{port}", "Referer": f"http://admin.archivebox.localhost:{port}/add/"}, + data={ + "url": recursive_test_site["root_url"], + "depth": "0", + "max_urls": "1", + "crawl_max_size": "0", + "snapshot_max_size": "0", + "main_plugins": ["wget"], + "tag": "restart-supervised-runner", + "url_filters_allowlist": r"127\.0\.0\.1[:/].*", + "url_filters_denylist": "", + "schedule": "", + "notes": "restart stopped supervised runner", + "persona": "Default", + "permissions": "public", + "start_paused": "", + "config": "{}", + "csrfmiddlewaretoken": csrf_token, + }, + timeout=10, + allow_redirects=False, + ) + assert response.status_code in (302, 303), response.text + + _wait_for_worker_state(tmp_path, "worker_runner", "RUNNING") + with use_archivebox_db(tmp_path): + crawl = Crawl.objects.order_by("-created_at").first() + assert crawl is not None + assert crawl.tags_str == "restart-supervised-runner" + assert crawl.urls == recursive_test_site["root_url"] + finally: + stop_server(tmp_path) + + +def _login_to_add_view(port: int) -> tuple[requests.Session, str]: + session = requests.Session() + wait_for_http(port, host=f"admin.archivebox.localhost:{port}", path="/admin/login/") + login_page = session.get( + f"http://127.0.0.1:{port}/admin/login/", + headers={"Host": f"admin.archivebox.localhost:{port}"}, + timeout=10, + ) + assert login_page.status_code == 200 + csrf_match = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', login_page.text) + assert csrf_match, login_page.text[:500] + login_response = session.post( + f"http://127.0.0.1:{port}/admin/login/", + headers={"Host": f"admin.archivebox.localhost:{port}", "Referer": f"http://admin.archivebox.localhost:{port}/admin/login/"}, + data={ + "username": "apitestadmin", + "password": "testpass123", + "csrfmiddlewaretoken": csrf_match.group(1), + "next": "/add/", + }, + timeout=10, + allow_redirects=False, + ) + assert login_response.status_code in (302, 303), login_response.text + add_page = wait_for_http(port, host=f"admin.archivebox.localhost:{port}", path="/add/") + assert add_page.status_code == 200 + add_csrf_match = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', add_page.text) + assert add_csrf_match, add_page.text[:500] + return session, add_csrf_match.group(1) + + +def _worker_state(cwd, worker_name: str) -> str | None: + script = f""" +import json +from archivebox.workers.supervisord_util import get_existing_supervisord_process, get_worker +supervisor = get_existing_supervisord_process() +worker = get_worker(supervisor, {worker_name!r}) if supervisor else None +print(json.dumps(worker)) +""" + stdout, stderr, returncode = run_archivebox_cmd_cwd(["manage", "shell", "-c", script], cwd=cwd, timeout=60) + assert returncode == 0, stderr or stdout + import json + + worker = json.loads(stdout.strip().splitlines()[-1]) + return worker.get("statename") if worker else None + + +def _stop_worker(cwd, worker_name: str) -> None: + script = f""" +from archivebox.workers.supervisord_util import get_existing_supervisord_process, stop_worker +supervisor = get_existing_supervisord_process() +assert supervisor is not None +stop_worker(supervisor, {worker_name!r}) +print("stopped") +""" + stdout, stderr, returncode = run_archivebox_cmd_cwd(["manage", "shell", "-c", script], cwd=cwd, timeout=60) + assert returncode == 0, stderr or stdout + + +def _wait_for_worker_state(cwd, worker_name: str, statename: str, timeout: int = 45) -> None: + deadline = time.time() + timeout + state = None + while time.time() < deadline: + state = _worker_state(cwd, worker_name) + if state == statename: + return + time.sleep(1) + raise AssertionError(f"Timed out waiting for {worker_name}={statename}, last state={state}") + + @pytest.mark.timeout(180) def test_add_view_post_creates_schedule_over_server(tmp_path, recursive_test_site): os.chdir(tmp_path)