The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-12 19:50:57 +05:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

fix: correlate snapshot completion leases

Browse Source
This commit is contained in:
Nick Sweeting 2026-09-02 14:18:00 -07:00
parent 8bc9f40e9f
commit 35e014ae90
No known key found for this signature in database
7 changed files with 130 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
archivebox/core/models.py
View File

@ -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}"

7
archivebox/misc/db.py
View File

@ -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):

19
archivebox/services/snapshot_service.py
View File

@ -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,

16
archivebox/tests/test_config_MAX_limits.py
View File

@ -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:

121
archivebox/tests/test_crawl_runner.py
View File

@ -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

8
archivebox/tests/test_snapshot_filesystem_migration.py
View File

@ -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):

24
archivebox/tests/test_snapshot_service.py
View File

@ -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