Preserve outputs during filesystem migration

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:

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.

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("<html>preserved output</html>")
     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"