From 35e014ae90a435c2fe7aa477bd91ebc96b5b8aca Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Wed, 2 Sep 2026 14:18:00 -0700 Subject: [PATCH] fix: correlate snapshot completion leases --- archivebox/core/models.py | 22 ++-- archivebox/misc/db.py | 7 +- archivebox/services/snapshot_service.py | 19 +-- archivebox/tests/test_config_MAX_limits.py | 16 ++- archivebox/tests/test_crawl_runner.py | 121 ++++++++++-------- .../test_snapshot_filesystem_migration.py | 8 +- archivebox/tests/test_snapshot_service.py | 24 ++++ 7 files changed, 130 insertions(+), 87 deletions(-) diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 86eee3d7..253451f5 100644 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -731,7 +731,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW if not plugin_names: return False retry_at = when or timezone.now() - for _attempt in range(3): + while True: current = type(self).objects.select_related("crawl").get(pk=self.pk) pending_plugins = {str(name).strip() for name in (current.config or {}).get("RETRY_PLUGINS", []) if str(name).strip()} config = {**(current.config or {}), "RETRY_PLUGINS": sorted(pending_plugins | set(plugin_names))} @@ -754,8 +754,6 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW if updated: crawl = current.crawl break - else: - return False self.refresh_from_db() if status in self.RUNNABLE_STATES and self.crawl_id: @@ -1063,21 +1061,12 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW if self._state.adding or update_fields is None or "notes" in update_fields: self.notes = sanitize_html_text(self.notes) - # The maintenance runner currently signals its explicit filesystem pass - # with this narrow update. Ordinary model saves must never move archive - # directories as an unrelated metadata side effect. - existing_snapshot = self.pk and not self._state.adding - maintenance_update = update_fields is not None and set(update_fields) == {"retry_at", "modified_at"} - if existing_snapshot and maintenance_update and self.fs_migration_needed: - self.migrate_filesystem_to_current_version() - super().save(*args, **kwargs) from django.db import transaction def finish_snapshot_save(): - self.remove_legacy_archive_symlink() - self.ensure_crawl_symlink() + self.reconcile_filesystem_links() crawl = Crawl.objects.filter(pk=self.crawl_id).first() if crawl is None: return @@ -1176,6 +1165,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW old_dir, new_dir = cleanup if not self._cleanup_old_migration_dir(old_dir, new_dir): raise SnapshotMigrationError(f"Could not clean up verified migration directory: {old_dir}") + self.reconcile_filesystem_links() return while current != target: @@ -1211,6 +1201,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW now = timezone.now() type(self).objects.filter(pk=self.pk).update(fs_version=target, modified_at=now) self.modified_at = now + self.reconcile_filesystem_links() def _fs_migrate_from_0_7_0_to_0_9_0(self, source_dir: Path | None = None, config: "ArchiveBoxBaseConfig | None" = None): return self._fs_migrate_legacy_to_0_9_0(source_dir=source_dir, config=config) @@ -2700,6 +2691,11 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW if points_to_current_path: legacy_path.unlink(missing_ok=True) + def reconcile_filesystem_links(self) -> None: + """Repair filesystem projections after a save or migration.""" + self.remove_legacy_archive_symlink() + self.ensure_crawl_symlink() + @cached_property def legacy_archive_path(self) -> str: return f"{CONSTANTS.ARCHIVE_DIR_NAME}/{self.timestamp}" diff --git a/archivebox/misc/db.py b/archivebox/misc/db.py index 1c07755a..d20e64dc 100644 --- a/archivebox/misc/db.py +++ b/archivebox/misc/db.py @@ -380,7 +380,10 @@ def log_sqlite_lock_holders(console: Any, *, db_path: Path = CONSTANTS.DATABASE_ def sqlite_lock_error(error: BaseException) -> bool: from django.db import OperationalError as DjangoOperationalError - return isinstance(error, (SQLiteOperationalError, DjangoOperationalError)) and "database is locked" in str(error).lower() + message = str(error).lower() + return isinstance(error, (SQLiteOperationalError, DjangoOperationalError)) and ( + "database is locked" in message or "database table is locked" in message + ) def retry_sqlite_locks(action: Callable[[], Any], *, label: str, stderr: TextIO | None = None) -> Any: @@ -392,7 +395,7 @@ def retry_sqlite_locks(action: Callable[[], Any], *, label: str, stderr: TextIO try: return action() except OperationalError as err: - if "database is locked" not in str(err).lower(): + if not sqlite_lock_error(err): raise except SQLiteOperationalError as err: if not sqlite_lock_error(err): diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index 9aa23c14..d13cd587 100644 --- a/archivebox/services/snapshot_service.py +++ b/archivebox/services/snapshot_service.py @@ -77,7 +77,7 @@ def finalize_completed_snapshot( # snapshot hook sequence (including cleanup) finished. ArchiveResult rows # are projections of that work, never prerequisites used to decide whether # the Snapshot may seal. - if not was_sealed and snapshot.status == Snapshot.StatusChoices.STARTED: + if not was_sealed and snapshot.status == Snapshot.StatusChoices.STARTED and snapshot.retry_at == owned_retry_at: snapshot.seal() snapshot.refresh_from_db() @@ -127,7 +127,7 @@ class SnapshotService(BaseService): def __init__(self, bus, *, crawl_id: str): self.crawl_id = crawl_id - self._run_ownership: dict[str, tuple[object, bool, list[str]]] = {} + self._run_ownership: dict[str, tuple[str, object, bool, list[str]]] = {} super().__init__(bus) self.bus.on(SnapshotEvent, self.on_SnapshotEvent) self.bus.on(SnapshotCompletedEvent, self.on_SnapshotCompletedEvent) @@ -149,18 +149,13 @@ 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.snapshot_id)] = (snapshot.retry_at, was_sealed, retry_plugins) + self._run_ownership[str(event.event_id)] = (str(event.snapshot_id), snapshot.retry_at, was_sealed, retry_plugins) async def on_SnapshotCompletedEvent(self, event: SnapshotCompletedEvent) -> None: - ownership = self._run_ownership.pop(str(event.snapshot_id), None) - if ownership is None: - from archivebox.core.models import Snapshot - - snapshot = await Snapshot.objects.only("status", "retry_at").filter(id=event.snapshot_id, crawl_id=self.crawl_id).afirst() - if snapshot is None: - return - ownership = (snapshot.retry_at, snapshot.status == Snapshot.StatusChoices.SEALED, []) - owned_retry_at, was_sealed, retry_plugins = ownership + ownership = self._run_ownership.pop(str(event.event_parent_id), None) + if ownership is None or ownership[0] != str(event.snapshot_id): + return + _, owned_retry_at, was_sealed, retry_plugins = ownership await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)( event.snapshot_id, owned_retry_at=owned_retry_at, diff --git a/archivebox/tests/test_config_MAX_limits.py b/archivebox/tests/test_config_MAX_limits.py index b8e2cd33..9ea445e2 100644 --- a/archivebox/tests/test_config_MAX_limits.py +++ b/archivebox/tests/test_config_MAX_limits.py @@ -195,7 +195,7 @@ def test_seal_snapshot_cancels_queued_descendants_after_crawl_max_size(): from archivebox.crawls.models import Crawl from archivebox.core.models import Snapshot from archivebox.services.snapshot_service import SnapshotService - from abx_dl.events import SnapshotCompletedEvent + from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent from abx_dl.orchestrator import create_bus crawl = Crawl.objects.create( @@ -231,17 +231,25 @@ def test_seal_snapshot_cancels_queued_descendants_after_crawl_max_size(): ) bus = create_bus(name=f"test_snapshot_limit_cancel_{str(crawl.id).replace('-', '_')}") - service = SnapshotService(bus, crawl_id=str(crawl.id)) + SnapshotService(bus, crawl_id=str(crawl.id)) try: async def emit_event() -> None: - await service.on_SnapshotCompletedEvent( - SnapshotCompletedEvent( + snapshot_event = bus.emit( + SnapshotEvent( url=root.url, snapshot_id=str(root.id), output_dir=str(root.output_dir), ), ) + await snapshot_event.now() + completed_event = SnapshotCompletedEvent( + url=root.url, + snapshot_id=str(root.id), + output_dir=str(root.output_dir), + ) + completed_event.event_parent_id = snapshot_event.event_id + await bus.emit(completed_event).now() asyncio.run(emit_event()) finally: diff --git a/archivebox/tests/test_crawl_runner.py b/archivebox/tests/test_crawl_runner.py index 8d541c3c..5bc55306 100644 --- a/archivebox/tests/test_crawl_runner.py +++ b/archivebox/tests/test_crawl_runner.py @@ -1084,60 +1084,13 @@ def test_crawl_completed_event_seals_finished_crawl(): assert crawl.retry_at is None -@pytest.mark.django_db(transaction=True) -def test_snapshot_completed_event_defers_finished_crawl_seal(): - from archivebox.base_models.models import get_or_create_system_user_pk - from archivebox.crawls.models import Crawl - from archivebox.core.models import Snapshot - from archivebox.services.snapshot_service import SnapshotService - from abx_dl.events import SnapshotCompletedEvent - from abx_dl.orchestrator import create_bus - from django.utils import timezone - - crawl = Crawl.objects.create( - urls="https://example.com", - created_by_id=get_or_create_system_user_pk(), - status=Crawl.StatusChoices.STARTED, - retry_at=timezone.now(), - ) - snapshot = Snapshot.objects.create( - url="https://example.com", - crawl=crawl, - status=Snapshot.StatusChoices.STARTED, - retry_at=None, - ) - - bus = create_bus(name=f"test_snapshot_completed_finished_crawl_{str(crawl.id).replace('-', '_')}") - service = SnapshotService(bus, crawl_id=str(crawl.id)) - try: - - async def emit_completed() -> None: - await service.on_SnapshotCompletedEvent( - SnapshotCompletedEvent( - url="https://example.com", - snapshot_id=str(snapshot.id), - output_dir=str(snapshot.output_dir), - ), - ) - - asyncio.run(emit_completed()) - finally: - asyncio.run(bus.destroy()) - - snapshot.refresh_from_db() - crawl.refresh_from_db() - assert snapshot.status == Snapshot.StatusChoices.SEALED - assert crawl.status == Crawl.StatusChoices.STARTED - assert crawl.retry_at is not None - - @pytest.mark.django_db(transaction=True) def test_snapshot_completed_event_bus_defers_finished_crawl_seal(): from archivebox.base_models.models import get_or_create_system_user_pk from archivebox.crawls.models import Crawl from archivebox.core.models import Snapshot from archivebox.services.snapshot_service import SnapshotService - from abx_dl.events import SnapshotCompletedEvent + from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent from abx_dl.orchestrator import create_bus from django.utils import timezone @@ -1155,20 +1108,25 @@ def test_snapshot_completed_event_bus_defers_finished_crawl_seal(): ) bus = create_bus(name=f"test_snapshot_completed_bus_finished_crawl_{str(crawl.id).replace('-', '_')}") - service = SnapshotService(bus, crawl_id=str(crawl.id)) - assert service is not None + SnapshotService(bus, crawl_id=str(crawl.id)) try: async def emit_completed() -> None: - emitted = bus.emit( - SnapshotCompletedEvent( + snapshot_event = bus.emit( + SnapshotEvent( url="https://example.com", snapshot_id=str(snapshot.id), output_dir=str(snapshot.output_dir), ), ) - await emitted.wait() - await emitted.event_results_list() + await snapshot_event.now() + completed_event = SnapshotCompletedEvent( + url="https://example.com", + snapshot_id=str(snapshot.id), + output_dir=str(snapshot.output_dir), + ) + completed_event.event_parent_id = snapshot_event.event_id + await bus.emit(completed_event).now() asyncio.run(emit_completed()) finally: @@ -1180,3 +1138,58 @@ def test_snapshot_completed_event_bus_defers_finished_crawl_seal(): assert snapshot.status == Snapshot.StatusChoices.SEALED assert crawl.status == Crawl.StatusChoices.STARTED assert crawl.retry_at is not None + + +@pytest.mark.django_db(transaction=True) +def test_delayed_snapshot_completion_cannot_seal_new_run(): + from datetime import timedelta + + from archivebox.base_models.models import get_or_create_system_user_pk + from archivebox.crawls.models import Crawl + from archivebox.core.models import Snapshot + from archivebox.services.snapshot_service import SnapshotService + from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent + from abx_dl.orchestrator import create_bus + from asgiref.sync import sync_to_async + from django.utils import timezone + + crawl = Crawl.objects.create(urls="https://example.com", created_by_id=get_or_create_system_user_pk()) + old_retry_at = timezone.now() + timedelta(minutes=5) + new_retry_at = old_retry_at + timedelta(minutes=5) + snapshot = Snapshot.objects.create( + url="https://example.com", + crawl=crawl, + status=Snapshot.StatusChoices.STARTED, + retry_at=old_retry_at, + ) + + bus = create_bus(name=f"test_delayed_snapshot_completion_{str(crawl.id).replace('-', '_')}") + SnapshotService(bus, crawl_id=str(crawl.id)) + try: + + async def emit_runs() -> None: + old_event = bus.emit( + SnapshotEvent(url=snapshot.url, snapshot_id=str(snapshot.id), output_dir=str(snapshot.output_dir)), + ) + await old_event.now() + await sync_to_async(Snapshot.objects.filter(pk=snapshot.pk).update, thread_sensitive=True)(retry_at=new_retry_at) + new_event = bus.emit( + SnapshotEvent(url=snapshot.url, snapshot_id=str(snapshot.id), output_dir=str(snapshot.output_dir)), + ) + await new_event.now() + completed_event = SnapshotCompletedEvent( + url=snapshot.url, + snapshot_id=str(snapshot.id), + output_dir=str(snapshot.output_dir), + ) + completed_event.event_parent_id = old_event.event_id + await bus.emit(completed_event).now() + + asyncio.run(emit_runs()) + finally: + asyncio.run(bus.wait_until_idle()) + asyncio.run(bus.destroy()) + + snapshot.refresh_from_db() + assert snapshot.status == Snapshot.StatusChoices.STARTED + assert snapshot.retry_at == new_retry_at diff --git a/archivebox/tests/test_snapshot_filesystem_migration.py b/archivebox/tests/test_snapshot_filesystem_migration.py index 95da354b..d057b237 100644 --- a/archivebox/tests/test_snapshot_filesystem_migration.py +++ b/archivebox/tests/test_snapshot_filesystem_migration.py @@ -34,15 +34,19 @@ def test_ordinary_snapshot_save_does_not_migrate_directories(snapshot): assert not (current_dir / "unknown" / "payload.bin").exists() -def test_maintenance_save_keeps_existing_runner_entrypoint(snapshot): +def test_filesystem_migration_repairs_crawl_link(snapshot): legacy_dir, current_dir = _make_legacy_snapshot(snapshot) + crawl_link = Path(snapshot.crawl.output_dir) / "snapshots" / Snapshot.extract_domain_from_url(snapshot.url) / str(snapshot.id) + crawl_link.unlink(missing_ok=True) - snapshot.save(update_fields=["retry_at", "modified_at"]) + snapshot.migrate_filesystem_to_current_version() snapshot.refresh_from_db() assert snapshot.fs_version == snapshot._fs_current_version() assert not legacy_dir.exists() assert (current_dir / "unknown" / "payload.bin").read_bytes() == b"filesystem migration payload\x00\xff" + assert crawl_link.is_symlink() + assert crawl_link.resolve() == current_dir.resolve() def test_filesystem_migration_resumes_after_shutdown_before_cleanup(snapshot, monkeypatch): diff --git a/archivebox/tests/test_snapshot_service.py b/archivebox/tests/test_snapshot_service.py index 9405f94c..ccc17ab3 100644 --- a/archivebox/tests/test_snapshot_service.py +++ b/archivebox/tests/test_snapshot_service.py @@ -1,7 +1,9 @@ import json +from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest +from django.db import close_old_connections from archivebox.core.models import ArchiveResult, Snapshot from archivebox.tests.conftest import run_archivebox_cmd @@ -74,6 +76,28 @@ def test_snapshot_completion_preserves_retry_scheduled_during_active_run(tmp_pat assert snapshot.config["RETRY_PLUGINS"] == ["title"] +def test_concurrent_plugin_scheduling_durably_merges_every_request(admin_user): + from archivebox.crawls.models import Crawl + + 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) + plugins = ["title", "wget", "screenshot", "pdf"] + + def schedule(plugin: str) -> bool: + close_old_connections() + try: + return Snapshot.objects.get(pk=snapshot.pk).schedule_plugin_run([plugin]) + finally: + close_old_connections() + + with ThreadPoolExecutor(max_workers=len(plugins)) as pool: + results = list(pool.map(schedule, plugins)) + + snapshot.refresh_from_db() + assert results == [True] * len(plugins) + assert snapshot.config["RETRY_PLUGINS"] == sorted(plugins) + + def test_snapshot_merge_consolidates_only_exact_hook_identity(admin_user): from archivebox.crawls.models import Crawl