Isolate queued snapshot deletion from the crawl scheduler

archivebox/cli/archivebox_run.py

@@ -294,6 +294,9 @@ def run_runner(
     )
     if daemon:
         os.environ[RUNNER_DAEMON_ENV] = "1"
+        from archivebox.workers.supervisord_util import SNAPSHOT_DELETE_WORKER, get_or_create_supervisord_process, start_worker
+
+        start_worker(get_or_create_supervisord_process(), SNAPSHOT_DELETE_WORKER())
 
     shutdown_state = None
     try:

archivebox/core/admin_snapshots.py

@@ -1543,11 +1543,11 @@ class SnapshotAdmin(SearchResultsAdminMixin, ConfigEditorMixin, BaseModelAdmin):
         permissions=["delete"],
     )
     def delete_snapshots(self, request, queryset):
-        """Keep the rows until the runner removes and verifies their files."""
+        """Keep the rows until the deletion worker removes and verifies their files."""
         total = queryset.update(status=Snapshot.DELETING_STATE, retry_at=RETRY_AT_MAX, modified_at=timezone.now())
         messages.success(
             request,
-            f"Queued {total} snapshots for deletion. They remain visible as Deleting until the runner verifies their files are removed.",
+            f"Queued {total} snapshots for deletion. They remain visible as Deleting until ArchiveBox verifies their files are removed.",
         )
 
     @admin.action(

archivebox/services/runner.py

@@ -1997,7 +1997,13 @@ def run_pending_crawls(
         raise_if_shutdown_requested()
         now_monotonic = time.monotonic()
         if crawl_id is None and now_monotonic - last_retention_at >= (60.0 if daemon else 1.0):
-            Snapshot.delete_requested(batch_size=100)
+            if not daemon:
+                from archivebox.workers.supervisord_util import get_existing_supervisord_process, get_worker
+
+                supervisor = get_existing_supervisord_process(quiet=True)
+                deletion_worker = get_worker(supervisor, "worker_snapshot_delete") if supervisor else None
+                if not deletion_worker or deletion_worker.get("statename") not in ("STARTING", "RUNNING"):
+                    Snapshot.delete_requested(batch_size=100)
             for model in (ArchiveResult, Snapshot, Crawl, Process):
                 # Keep the tight scheduler loop anchored on indexed delete_at
                 # columns only. Backfilling missing delete_at values has to read

archivebox/tests/test_cli_server.py

@@ -448,7 +448,11 @@ def test_port_compatibility_does_not_add_listeners_outside_docker_defaults(tmp_p
         check=True,
     )
     workers = json.loads(result.stdout)
-    assert [worker["name"] for worker in workers] == ["worker_runserver" if debug else "worker_daphne", "worker_runner"]
+    assert [worker["name"] for worker in workers] == [
+        "worker_runserver" if debug else "worker_daphne",
+        "worker_runner",
+        "worker_snapshot_delete",
+    ]
     command = shlex.split(workers[0]["command"])
     assert "--endpoint" not in command
     assert (f"127.0.0.1:{port}" if debug else f"--port={port}") in command

archivebox/tests/test_snapshot_delete_worker.py

@@ -0,0 +1,114 @@
+"""Exercise filesystem failures and worker isolation with a real server."""
+
+import json
+import os
+import signal
+import time
+from pathlib import Path
+
+import pytest
+
+from archivebox.tests.conftest import (
+    cli_env,
+    get_free_port,
+    kill_processes_for_data_dir,
+    run_archivebox_cmd,
+    start_archivebox_server,
+    stop_archivebox_process,
+    wait_for_log,
+)
+from archivebox.tests.test_orm_helpers import use_archivebox_db
+
+
+@pytest.mark.django_db(transaction=True)
+def test_failed_and_paused_deletion_worker_does_not_block_captures(initialized_archive, recursive_test_site):
+    from archivebox.core.models import ArchiveResult, Snapshot
+    from archivebox.machine.models import Process
+
+    port = get_free_port()
+    env = cli_env(
+        live=True,
+        server=True,
+        port=port,
+        PLUGINS="title",
+        SAVE_TITLE="True",
+        SEARCH_BACKEND_ENGINE="ripgrep",
+    )
+    prepared = run_archivebox_cmd(
+        [
+            "manage",
+            "shell",
+            "-c",
+            "import json; from archivebox.crawls.models import Crawl; "
+            "from archivebox.core.models import Snapshot, ArchiveResult; "
+            "from archivebox.base_models.models import get_or_create_system_user_pk; "
+            "c=Crawl.objects.create(urls='https://example.com/delete',created_by_id=get_or_create_system_user_pk(),status='sealed'); "
+            "s=Snapshot.objects.create(url='https://example.com/delete',crawl=c,status='deleting'); "
+            "p=s.output_dir/'wget'; p.mkdir(parents=True,exist_ok=True); "
+            "(p/'index.html').write_text('keep this until cleanup succeeds'); "
+            "ArchiveResult.objects.create(snapshot=s,plugin='wget',hook_name='on_Snapshot__06_wget',status='succeeded'); "
+            "p.chmod(0o500); print(json.dumps({'id':str(s.pk),'path':str(p)}))",
+        ],
+        cwd=initialized_archive,
+        env=env,
+    )
+    assert prepared.returncode == 0, prepared.stderr or prepared.stdout
+    fixture = json.loads(prepared.stdout.strip().splitlines()[-1])
+    protected = Path(fixture["path"])
+    worker_log = initialized_archive / "logs" / "worker_snapshot_delete.log"
+    server = None
+    deletion_pid = None
+    try:
+        server = start_archivebox_server(initialized_archive, port=port, env=env, log_name="delete-worker-server.log")
+        wait_for_log(worker_log, "PermissionError", timeout=30)
+        with use_archivebox_db(initialized_archive):
+            deletion = Process.objects.get(worker_type="worker_snapshot_delete", status="running")
+            runner = Process.objects.get(worker_type="worker_runner", status="running")
+            assert deletion.pid != runner.pid
+            deletion_pid = deletion.pid
+            assert Snapshot.objects.filter(pk=fixture["id"], status="deleting").exists()
+            assert ArchiveResult.objects.filter(snapshot_id=fixture["id"]).exists()
+        assert (protected / "index.html").is_file()
+        # Suspend the actual cleanup process. The scheduler and its real title
+        # hook must still finish a new capture without waiting for this worker.
+        os.kill(deletion_pid, signal.SIGSTOP)
+        added = run_archivebox_cmd(
+            ["add", "--bg", "--plugins=title", recursive_test_site["root_url"]],
+            cwd=initialized_archive,
+            env=env,
+        )
+        assert added.returncode == 0, added.stderr or added.stdout
+        deadline = time.monotonic() + 60
+        with use_archivebox_db(initialized_archive):
+            while time.monotonic() < deadline:
+                captured = Snapshot.objects.filter(url=recursive_test_site["root_url"], status="sealed").first()
+                if captured is not None:
+                    break
+                assert server.poll() is None
+                time.sleep(0.1)
+            else:
+                pytest.fail((initialized_archive / "logs" / "worker_runner.log").read_text()[-12000:])
+            result = ArchiveResult.objects.get(snapshot=captured, plugin="title", status="succeeded")
+            assert result.output_files
+            assert captured.title == "Root"
+            assert Snapshot.objects.filter(pk=fixture["id"], status="deleting").exists()
+        protected.chmod(0o700)
+        os.kill(deletion_pid, signal.SIGCONT)
+        deadline = time.monotonic() + 75
+        with use_archivebox_db(initialized_archive):
+            while Snapshot.objects.filter(pk=fixture["id"]).exists() and time.monotonic() < deadline:
+                time.sleep(0.1)
+            assert not Snapshot.objects.filter(pk=fixture["id"]).exists(), worker_log.read_text()
+            assert not ArchiveResult.objects.filter(snapshot_id=fixture["id"]).exists()
+        assert not protected.parent.exists()
+    finally:
+        if protected.exists():
+            protected.chmod(0o700)
+        if deletion_pid is not None:
+            try:
+                os.kill(deletion_pid, signal.SIGCONT)
+            except ProcessLookupError:
+                pass
+        if server is not None:
+            stop_archivebox_process(server)
+        kill_processes_for_data_dir(initialized_archive)

archivebox/workers/management/commands/snapshot_delete.py

@@ -0,0 +1,23 @@
+import time
+
+from django.core.management.base import BaseCommand
+
+
+class Command(BaseCommand):
+    help = "Remove queued Snapshot files before deleting their database rows."
+
+    def handle(self, *args, **kwargs):
+        from archivebox.config import CONSTANTS
+        from archivebox.core.models import Snapshot
+        from archivebox.machine.models import Process
+
+        current = Process.current()
+        current.mark_running(
+            process_type=Process.TypeChoices.WORKER,
+            worker_type="worker_snapshot_delete",
+            pwd=str(CONSTANTS.DATA_DIR),
+        )
+        while True:
+            Snapshot.delete_requested(batch_size=100)
+            current.heartbeat()
+            time.sleep(60)

archivebox/workers/supervisord_util.py

@@ -301,6 +301,22 @@ def RUNNER_WORKER():
     }
 
 
+def SNAPSHOT_DELETE_WORKER():
+    command = archivebox_cmd("manage", "snapshot_delete")
+    return {
+        "name": "worker_snapshot_delete",
+        "command": _shell_join(command),
+        "autostart": "false",
+        "autorestart": "true",
+        "environment": _archivebox_worker_environment(command, PYTHONUNBUFFERED="1"),
+        "stopasgroup": "true",
+        "killasgroup": "true",
+        "stopwaitsecs": "30",
+        "stdout_logfile": "logs/worker_snapshot_delete.log",
+        "redirect_stderr": "true",
+    }
+
+
 def RUNNER_ONCE_WORKER(args, name="worker_runner_once"):
     command = archivebox_cmd("run", "--no-stdin", *args)
     return {
@@ -1336,6 +1352,9 @@ def build_server_worker_plan(*, config, host: str, port: str, debug: bool, reloa
         bg_workers = [(RUNNER_WORKER(), False)]
         log_files = ["logs/worker_daphne.log", "logs/worker_runner.log"]
 
+    bg_workers.append((SNAPSHOT_DELETE_WORKER(), False))
+    log_files.append("logs/worker_snapshot_delete.log")
+
     sonic_worker = get_sonic_supervisord_worker_from_plugin(config)
     if sonic_worker is not None:
         try: