From ba2bc83809c38758aaf592df52d663fac0cf5a5b Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Wed, 12 Aug 2026 22:07:34 -0700 Subject: [PATCH] Preserve outputs during filesystem migration --- archivebox/core/models.py | 65 ++++++++++++------- archivebox/services/runner.py | 50 -------------- .../test_cli_update_reindex_snapshots.py | 16 ++++- 3 files changed, 56 insertions(+), 75 deletions(-) diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 7fddead9..017afa1a 100644 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -1130,9 +1130,9 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW cleanup = migration(source_dir=source_dir, config=runtime_config) or cleanup current = next_ver + self.fs_version = current source_dir = None - self.fs_version = target if cleanup: self._pending_fs_migration_cleanup = cleanup @@ -1179,6 +1179,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW new_dir = Path(target_dir) if target_dir else self.get_storage_path_for_version("0.9.0") if old_dir == new_dir: + self.convert_index_json_to_jsonl(output_dir=new_dir) + self.hydrate_archiveresult_output_metadata(snapshot_dir=new_dir) return None if old_dir.is_symlink(): @@ -1916,31 +1918,50 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW index_path = output_dir / CONSTANTS.JSONL_INDEX_FILENAME index_path.parent.mkdir(parents=True, exist_ok=True) - # Track unique binaries and processes to avoid duplicates + archive_results = list(self.archiveresult_set.select_related("process__binary").order_by("start_ts")) + + # Build canonical records before replacing the file so legacy records + # without a corresponding DB row can be retained byte-for-byte. binaries_seen = set() processes_seen = set() + records = [self.to_json()] + for ar in archive_results: + process = ar.process_record + if process and process.binary and process.binary_id not in binaries_seen: + binaries_seen.add(process.binary_id) + records.append(process.binary.to_json()) + if process and process.id not in processes_seen: + processes_seen.add(process.id) + records.append(process.to_json()) + records.append(ar.to_json(snapshot_output_dir=output_dir)) + + canonical_keys = { + (record.get("type"), str(record.get("id"))) for record in records if record.get("type") and record.get("id") is not None + } + preserved_lines = [] + if index_path.exists(): + for line in index_path.read_text(encoding="utf-8").splitlines(keepends=True): + try: + existing_record = json.loads(line) + except json.JSONDecodeError: + preserved_lines.append(line) + continue + if existing_record.get("type") == "Snapshot": + records[0] = {**existing_record, **records[0]} + continue + key = (existing_record.get("type"), str(existing_record.get("id"))) + if existing_record.get("id") is None or key not in canonical_keys: + preserved_lines.append(line) tmp_index_path = index_path.with_name(f".{index_path.name}.tmp") - with open(tmp_index_path, "w") as f: - # Write Snapshot record first (to_json includes crawl_id, fs_version) - f.write(json.dumps(self.to_json()) + "\n") - - # Write ArchiveResult records with their associated Binary and Process - # Use select_related to optimize queries - for ar in self.archiveresult_set.select_related("process__binary").order_by("start_ts"): - process = ar.process_record - # Write Binary record if not already written - if process and process.binary and process.binary_id not in binaries_seen: - binaries_seen.add(process.binary_id) - f.write(json.dumps(process.binary.to_json()) + "\n") - - # Write Process record if not already written - if process and process.id not in processes_seen: - processes_seen.add(process.id) - f.write(json.dumps(process.to_json()) + "\n") - - # Write ArchiveResult record - f.write(json.dumps(ar.to_json(snapshot_output_dir=output_dir)) + "\n") + with open(tmp_index_path, "w", encoding="utf-8") as f: + f.write(json.dumps(records[0]) + "\n") + for line in preserved_lines: + f.write(line) + if not line.endswith("\n"): + f.write("\n") + for record in records[1:]: + f.write(json.dumps(record) + "\n") os.replace(tmp_index_path, index_path) def read_index_jsonl(self, output_dir: Path | None = None) -> dict: diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 624fa658..da7c2c7f 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -2163,53 +2163,6 @@ def _run_due_binary() -> bool: return True -def _fast_forward_same_path_snapshot_fs_versions(batch_size: int = 10000) -> bool: - from django.db import connection - - from archivebox.core.models import Snapshot, ArchiveResult - - now = timezone.now() - current_version = Snapshot._fs_current_version() - same_path_versions = ("0.9.0", "0.9.1", "0.9.2", "0.9.3") - with connection.cursor() as cursor: - cursor.execute( - """ - UPDATE core_snapshot - SET fs_version = %s, - retry_at = CASE - WHEN EXISTS ( - SELECT 1 - FROM core_archiveresult - WHERE core_archiveresult.snapshot_id = core_snapshot.id - AND core_archiveresult.status = %s - ) - THEN retry_at - ELSE NULL - END, - modified_at = %s - WHERE id IN ( - SELECT id - FROM core_snapshot - WHERE status = %s - AND retry_at <= %s - AND fs_version IN (%s, %s, %s, %s) - ORDER BY retry_at, created_at - LIMIT %s - ) - """, - [ - current_version, - ArchiveResult.StatusChoices.QUEUED, - now, - Snapshot.StatusChoices.SEALED, - now, - *same_path_versions, - batch_size, - ], - ) - return bool(cursor.rowcount) - - def run_pending_crawls( *, daemon: bool = False, @@ -2253,9 +2206,6 @@ def run_pending_crawls( if schedule.is_due(now): schedule.enqueue(queued_at=now) - if _fast_forward_same_path_snapshot_fs_versions(): - continue - if maintenance_only: # Filesystem migration is independent of lifecycle status; do not # tick queued snapshots or start their extraction work here. diff --git a/archivebox/tests/test_cli_update_reindex_snapshots.py b/archivebox/tests/test_cli_update_reindex_snapshots.py index c245f349..62545e71 100644 --- a/archivebox/tests/test_cli_update_reindex_snapshots.py +++ b/archivebox/tests/test_cli_update_reindex_snapshots.py @@ -6,7 +6,7 @@ from archivebox.tests.conftest import cli_env, run_archivebox_cmd import pytest from django.utils import timezone -from archivebox.core.models import Snapshot +from archivebox.core.models import ArchiveResult, Snapshot from archivebox.tests.migrations_helpers import filesystem_manifest from archivebox.tests.test_orm_helpers import use_archivebox_db @@ -127,7 +127,7 @@ def test_update_imports_orphaned_snapshots(tmp_path, initialized_archive): migrated_dir = legacy_dir.resolve() assert migrated_dir.exists() - assert (migrated_dir / "index.jsonl").exists() + assert '{"type":"Process","id":"incomplete"}\n' in (migrated_dir / "index.jsonl").read_text() assert (migrated_dir / "singlefile.html").exists() @@ -149,6 +149,12 @@ def test_update_migrates_every_declared_filesystem_version(tmp_path, initialized ) snapshot.refresh_from_db() source_dir = tmp_path / "archive" / snapshot.timestamp if legacy_layout else destination + result = ArchiveResult.objects.create( + snapshot=snapshot, + plugin="singlefile", + hook_name="on_Snapshot__singlefile", + status=ArchiveResult.StatusChoices.SUCCEEDED, + ) if legacy_layout: source_dir.mkdir(parents=True, exist_ok=True) @@ -172,6 +178,8 @@ def test_update_migrates_every_declared_filesystem_version(tmp_path, initialized (source_dir / "unknown" / "empty").mkdir(parents=True, exist_ok=True) (source_dir / "unknown" / "payload.bin").write_bytes(b"filesystem migration payload\x00\xff") (source_dir / "unknown" / "payload-link").symlink_to("payload.bin") + (source_dir / "singlefile").mkdir(exist_ok=True) + (source_dir / "singlefile" / "singlefile.html").write_text("preserved output") original_tree = filesystem_manifest(source_dir) original_tree.pop("index.jsonl", None) @@ -185,7 +193,9 @@ def test_update_migrates_every_declared_filesystem_version(tmp_path, initialized migrated_tree = filesystem_manifest(migrated_dir) assert snapshot.status == Snapshot.StatusChoices.QUEUED assert snapshot.retry_at is not None - assert snapshot.archiveresult_set.count() == 0 + result.refresh_from_db() + assert result.output_files + assert result.output_size > 0 if legacy_layout: assert (migrated_dir / "existing-user-output.bin").read_bytes() == b"preserve interrupted migration output"