Make API pause win over runner leases

archivebox/core/models.py

@@ -3644,9 +3644,12 @@ class SnapshotMachine(BaseStateMachine):
 
     @paused.enter
     def enter_paused(self):
-        self.snapshot.update_and_requeue(
-            retry_at=RETRY_AT_MAX,
-            status=Snapshot.StatusChoices.PAUSED,
+        self.snapshot.safe_update(
+            {
+                "retry_at": RETRY_AT_MAX,
+                "status": Snapshot.StatusChoices.PAUSED,
+            },
+            extra_filter={"status__in": Snapshot.RUNNABLE_STATES},
         )
 
     @started.enter

archivebox/crawls/models.py

@@ -1647,11 +1647,15 @@ class CrawlMachine(BaseStateMachine):
 
     @paused.enter
     def enter_paused(self):
-        self.crawl.update_and_requeue(
-            retry_at=RETRY_AT_MAX,
-            status=Crawl.StatusChoices.PAUSED,
+        paused = self.crawl.safe_update(
+            {
+                "retry_at": RETRY_AT_MAX,
+                "status": Crawl.StatusChoices.PAUSED,
+            },
+            extra_filter={"status__in": Crawl.RUNNABLE_STATES},
         )
-        self.crawl.schedule_child_snapshots_for_pause()
+        if paused:
+            self.crawl.schedule_child_snapshots_for_pause()
 
     @sealed.enter
     def enter_sealed(self):

archivebox/tests/test_api_v1_core_snapshot_snapshot_id.py

@@ -1,4 +1,5 @@
 import json
+from datetime import timedelta
 from pathlib import Path
 from threading import Thread
 
@@ -98,6 +99,19 @@ def test_basic_success_case_request(client, tmp_path, api_admin_user, api_header
     assert response.status_code == 200, response.content
 
 
+def test_snapshot_pause_wins_over_concurrent_runner_lease(api_admin_user):
+    crawl = Crawl.objects.create(urls="https://example.com/snapshot-pause-race", created_by=api_admin_user)
+    snapshot = Snapshot.objects.create(url="https://example.com/snapshot-pause-race", crawl=crawl)
+    claimed_until = timezone.now() + timedelta(seconds=60)
+    Snapshot.objects.filter(pk=snapshot.pk).update(retry_at=claimed_until)
+
+    assert snapshot.pause() is True
+
+    snapshot.refresh_from_db()
+    assert snapshot.status == Snapshot.StatusChoices.PAUSED
+    assert snapshot.retry_at == RETRY_AT_MAX
+
+
 def test_snapshot_pause_resume_api_cascades_active_archiveresults_and_preserves_finished_rows(
     request,
     tmp_path,

archivebox/tests/test_api_v1_crawls_crawl_crawl_id.py

@@ -114,6 +114,18 @@ def test_basic_success_case_request(client, tmp_path, api_admin_user, api_header
     assert response.status_code == 200, response.content
 
 
+def test_crawl_pause_wins_over_concurrent_runner_lease(api_admin_user):
+    crawl = Crawl.objects.create(urls="https://example.com/crawl-pause-race", created_by=api_admin_user)
+    claimed_until = timezone.now() + timedelta(seconds=60)
+    Crawl.objects.filter(pk=crawl.pk).update(retry_at=claimed_until)
+
+    assert crawl.pause() is True
+
+    crawl.refresh_from_db()
+    assert crawl.status == Crawl.StatusChoices.PAUSED
+    assert crawl.retry_at == RETRY_AT_MAX
+
+
 def test_crawl_pause_resume_api_cascades_archiveresults_and_leaves_finished_snapshot_results_alone(
     request,
     tmp_path,