From e878007b6a1cd79e700514aaa2022d2125693090 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Wed, 2 Sep 2026 14:35:57 -0700 Subject: [PATCH] fix: bound snapshot scheduling ownership --- archivebox/core/models.py | 4 +++- archivebox/misc/db.py | 13 ++++++++----- archivebox/services/runner.py | 15 +-------------- archivebox/services/snapshot_service.py | 8 +++++--- archivebox/tests/test_snapshot_service.py | 4 ++++ 5 files changed, 21 insertions(+), 23 deletions(-) diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 253451f5..e37dbda7 100644 --- a/archivebox/core/models.py +++ b/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: diff --git a/archivebox/misc/db.py b/archivebox/misc/db.py index d20e64dc..c5cd9262 100644 --- a/archivebox/misc/db.py +++ b/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 diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index ee24874f..83330923 100644 --- a/archivebox/services/runner.py +++ b/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 diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index d13cd587..4d9fbeb0 100644 --- a/archivebox/services/snapshot_service.py +++ b/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, diff --git a/archivebox/tests/test_snapshot_service.py b/archivebox/tests/test_snapshot_service.py index ccc17ab3..f5c3e041 100644 --- a/archivebox/tests/test_snapshot_service.py +++ b/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)