fix: bound snapshot scheduling ownership

archivebox/core/models.py

@@ -273,7 +273,7 @@ class SnapshotQuerySet(models.QuerySet):
 
         last_values = None
         value_field_names = tuple(dict.fromkeys([*ordered_field_names, pk_field]))
-        while True:
+        for _attempt in range(8):
             batch_qs = self.order_by(*ordering)
             if last_values is not None:
                 page_filter = models.Q()
@@ -754,6 +754,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             if updated:
                 crawl = current.crawl
                 break
+        else:
+            raise RuntimeError(f"Snapshot {self.pk} changed repeatedly while scheduling plugins")
 
         self.refresh_from_db()
         if status in self.RUNNABLE_STATES and self.crawl_id:

archivebox/misc/db.py

@@ -387,25 +387,28 @@ def sqlite_lock_error(error: BaseException) -> bool:
 
 
 def retry_sqlite_locks(action: Callable[[], Any], *, label: str, stderr: TextIO | None = None) -> Any:
+    from django.conf import settings
     from django.db import OperationalError, connections
     from rich.console import Console
 
     console = Console(file=stderr or None, stderr=stderr is None)
+    started_at = time.monotonic()
+    retry_timeout = settings.CONFIG.SQLITE_LOCK_RETRY_TIMEOUT
+    retry_interval = settings.CONFIG.SQLITE_LOCK_RETRY_INTERVAL
     while True:
         try:
             return action()
-        except OperationalError as err:
+        except (OperationalError, SQLiteOperationalError) as err:
             if not sqlite_lock_error(err):
                 raise
-        except SQLiteOperationalError as err:
-            if not sqlite_lock_error(err):
+            if retry_timeout and time.monotonic() - started_at >= retry_timeout:
                 raise
 
         connections.close_all()
-        console.print(f"[yellow][*] SQLite database is locked while {label}; retrying in 5s...[/yellow]")
+        console.print(f"[yellow][*] SQLite database is locked while {label}; retrying in {retry_interval:g}s...[/yellow]")
         log_sqlite_lock_holders(console)
         with console.status("[yellow]Waiting for SQLite database lock to clear...[/yellow]", spinner="dots"):
-            time.sleep(5.0)
+            time.sleep(retry_interval)
 
 
 @contextmanager

archivebox/services/runner.py

@@ -75,7 +75,7 @@ from .binary_service import ArchiveBoxBinaryService, project_abxpkg_derived_cach
 from .crawl_service import CrawlService
 from .machine_service import MachineService
 from .process_service import ProcessService as PersistedProcessService
-from .snapshot_service import SnapshotService, finalize_completed_snapshot, project_discovered_snapshots
+from .snapshot_service import SnapshotService, project_discovered_snapshots
 from .tag_service import TagService
 
 
@@ -1026,19 +1026,6 @@ class CrawlRunner:
                     raise RuntimeError(f"Snapshot {snapshot_id} did not complete")
                 await completed_snapshot.wait(timeout=snapshot_phase_timeout)
                 await completed_snapshot.event_results_list()
-                # SnapshotCompletedEvent is the normal projection path, but the
-                # runner is the scheduler owner. Finalize idempotently here too
-                # so a completed snapshot cannot remain STARTED if the event was
-                # observed before its DB projector advanced the lifecycle.
-                crawl_limit_stop_reason = CrawlLimitState.from_config(config).get_stop_reason()
-                await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)(
-                    snapshot_id,
-                    owned_retry_at=snapshot["retry_at"],
-                    was_sealed=snapshot["status"] == "sealed",
-                    consumed_retry_plugins=snapshot["retry_plugins"],
-                    output_dir=output_dir,
-                    crawl_limit_stop_reason=crawl_limit_stop_reason,
-                )
                 if snapshot["status"] == "sealed":
                     await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir)
                     return

archivebox/services/snapshot_service.py

@@ -149,12 +149,14 @@ class SnapshotService(BaseService):
             if snapshot.status == Snapshot.StatusChoices.STARTED:
                 await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)()
             retry_plugins = [str(name).strip() for name in (snapshot.config or {}).get("RETRY_PLUGINS", []) if str(name).strip()]
-            self._run_ownership[str(event.event_id)] = (str(event.snapshot_id), snapshot.retry_at, was_sealed, retry_plugins)
+            self._run_ownership[str(event.snapshot_id)] = (str(event.event_id), snapshot.retry_at, was_sealed, retry_plugins)
 
     async def on_SnapshotCompletedEvent(self, event: SnapshotCompletedEvent) -> None:
-        ownership = self._run_ownership.pop(str(event.event_parent_id), None)
-        if ownership is None or ownership[0] != str(event.snapshot_id):
+        snapshot_id = str(event.snapshot_id)
+        ownership = self._run_ownership.get(snapshot_id)
+        if ownership is None or ownership[0] != str(event.event_parent_id):
             return
+        self._run_ownership.pop(snapshot_id, None)
         _, owned_retry_at, was_sealed, retry_plugins = ownership
         await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)(
             event.snapshot_id,

archivebox/tests/test_snapshot_service.py

@@ -78,6 +78,10 @@ def test_snapshot_completion_preserves_retry_scheduled_during_active_run(tmp_pat
 
 def test_concurrent_plugin_scheduling_durably_merges_every_request(admin_user):
     from archivebox.crawls.models import Crawl
+    from django.db import connection
+
+    if connection.vendor != "sqlite":
+        pytest.skip("exercises SQLite concurrent-writer scheduling")
 
     crawl = Crawl.objects.create(urls="https://example.com/plugin-race", created_by=admin_user)
     snapshot = Snapshot.objects.create(url="https://example.com/plugin-race", crawl=crawl)