From 06fbed2be2cedf879d97648aed8a78a8a2f7c27e Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Sat, 1 Aug 2026 13:31:52 -0700 Subject: [PATCH] Make Process metadata migration transactional --- .../0027_copy_archiveresult_to_process.py | 131 ++++++------ .../tests/test_migration_process_integrity.py | 199 ++++++++++++++++++ 2 files changed, 267 insertions(+), 63 deletions(-) create mode 100644 archivebox/tests/test_migration_process_integrity.py diff --git a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py index a564b252..f0a5c1f4 100644 --- a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py +++ b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py @@ -1,7 +1,7 @@ # Generated by hand on 2026-01-01 # Copies ArchiveResult cmd/pwd/cmd_version data to Process records before removing old fields -from django.db import migrations, connection +from django.db import migrations import json from pathlib import Path from archivebox.uuid_compat import uuid7 @@ -283,7 +283,7 @@ def copy_archiveresult_data_to_process(apps, schema_editor): if schema_editor.connection.vendor != "sqlite": return - cursor = connection.cursor() + cursor = schema_editor.connection.cursor() # Check if old fields still exist (skip if fresh install or already migrated) cursor.execute("PRAGMA table_info(core_archiveresult)") @@ -295,96 +295,101 @@ def copy_archiveresult_data_to_process(apps, schema_editor): # Check if process_id field exists (should exist from 0026) if "process_id" not in cols: - print("✗ ERROR: process_id field not found. Migration 0026 must run first.") - return + raise RuntimeError("process_id field not found. Migration 0026 must run first.") # Get or create Machine.current() machine_id = get_or_create_current_machine(cursor) - # Get ArchiveResults without process_id that have cmd data + # Get every ArchiveResult without a Process. Rows with empty execution + # metadata still need a Process before the legacy columns can be removed. # Use plugin (extractor was renamed to plugin in migration 0025) cursor.execute(""" SELECT id, snapshot_id, plugin, cmd, pwd, cmd_version, status, start_ts, end_ts, created_at FROM core_archiveresult WHERE process_id IS NULL - AND (cmd IS NOT NULL OR pwd IS NOT NULL) """) results = cursor.fetchall() if not results: print(" ✓ No ArchiveResults need Process migration") - return - - print(f" - Migrating {len(results)} ArchiveResults to Process rows...") + else: + print(f" - Migrating {len(results)} ArchiveResults to Process rows...") migrated_count = 0 - skipped_count = 0 - error_count = 0 - for i, row in enumerate(results): ar_id, snapshot_id, plugin, cmd_raw, pwd, cmd_version, status, start_ts, end_ts, created_at = row - try: - # Parse cmd field - cmd_array = parse_cmd_field(cmd_raw) + # Parse cmd field + cmd_array = parse_cmd_field(cmd_raw) - # Extract binary info from cmd[0] if available - binary_id = None - if cmd_array and cmd_array[0]: - binary_name = Path(cmd_array[0]).name or plugin # Fallback to plugin name - binary_abspath = cmd_array[0] - binary_version = cmd_version or "" + # Extract binary info from cmd[0] if available + binary_id = None + if cmd_array and cmd_array[0]: + binary_name = Path(cmd_array[0]).name or plugin # Fallback to plugin name + binary_abspath = cmd_array[0] + binary_version = cmd_version or "" - # Get or create Binary record - binary_id = get_or_create_binary( - cursor, - machine_id, - binary_name, - binary_abspath, - binary_version, - ) - - # Map status - process_status, exit_code = map_status(status) - - # Set timestamps - started_at = start_ts or created_at - ended_at = end_ts if process_status == "exited" else None - - # Create Process record - process_id = create_process( - cursor=cursor, - machine_id=machine_id, - pwd=pwd or "", - cmd=cmd_array, - status=process_status, - exit_code=exit_code, - started_at=started_at, - ended_at=ended_at, - binary_id=binary_id, + # Get or create Binary record + binary_id = get_or_create_binary( + cursor, + machine_id, + binary_name, + binary_abspath, + binary_version, ) + elif cmd_version: + raise RuntimeError(f"ArchiveResult {ar_id} has cmd_version metadata but no command") - # Link ArchiveResult to Process - cursor.execute( - "UPDATE core_archiveresult SET process_id = ? WHERE id = ?", - [process_id, ar_id], - ) + # Map status + process_status, exit_code = map_status(status) - migrated_count += 1 - if migrated_count % PROGRESS_EVERY == 0: - print(f" migrated {migrated_count}/{len(results)} ArchiveResults...") + # Set timestamps + started_at = start_ts or created_at + ended_at = end_ts if process_status == "exited" else None - except Exception as e: - print(f"✗ Error migrating ArchiveResult {ar_id}: {e}") - import traceback + # Create Process record + process_id = create_process( + cursor=cursor, + machine_id=machine_id, + pwd=pwd or "", + cmd=cmd_array, + status=process_status, + exit_code=exit_code, + started_at=started_at, + ended_at=ended_at, + binary_id=binary_id, + ) - traceback.print_exc() - error_count += 1 - continue + # Link ArchiveResult to Process + cursor.execute( + "UPDATE core_archiveresult SET process_id = ? WHERE id = ?", + [process_id, ar_id], + ) + if cursor.rowcount != 1: + raise RuntimeError(f"ArchiveResult {ar_id} was not linked to Process {process_id}") - print(f" ✓ Process migration complete: {migrated_count} migrated, {skipped_count} skipped, {error_count} errors") + migrated_count += 1 + if migrated_count % PROGRESS_EVERY == 0: + print(f" migrated {migrated_count}/{len(results)} ArchiveResults...") + + cursor.execute(""" + SELECT ar.id, ar.cmd, ar.pwd, ar.cmd_version, ar.process_id, + process.cmd, process.pwd, binary.version + FROM core_archiveresult AS ar + LEFT JOIN machine_process AS process ON process.id = ar.process_id + LEFT JOIN machine_binary AS binary ON binary.id = process.binary_id + """) + for ar_id, cmd_raw, pwd, cmd_version, process_id, process_cmd_raw, process_pwd, binary_version in cursor.fetchall(): + if not process_id: + raise RuntimeError(f"ArchiveResult {ar_id} is not linked to a Process") + if parse_cmd_field(process_cmd_raw) != parse_cmd_field(cmd_raw) or process_pwd != (pwd or ""): + raise RuntimeError(f"ArchiveResult {ar_id} execution metadata was not preserved") + if cmd_version and binary_version != cmd_version: + raise RuntimeError(f"ArchiveResult {ar_id} command version was not preserved") + + print(f" ✓ Process migration complete: {migrated_count} migrated") class Migration(migrations.Migration): diff --git a/archivebox/tests/test_migration_process_integrity.py b/archivebox/tests/test_migration_process_integrity.py new file mode 100644 index 00000000..7ef3158b --- /dev/null +++ b/archivebox/tests/test_migration_process_integrity.py @@ -0,0 +1,199 @@ +import json +import os +import sqlite3 +import subprocess + + +MIGRATION_0026 = "0026_add_process_to_archiveresult" +MIGRATION_0027 = "0027_copy_archiveresult_to_process" +NOW = "2024-01-01 12:00:00" +USER_ID = 42 +CRAWL_ID = "10000000000000000000000000000000" +SNAPSHOT_ID = "20000000000000000000000000000000" + + +def run_migration(data_dir, target): + env = os.environ.copy() + env.update( + { + "DATA_DIR": str(data_dir), + "USE_COLOR": "False", + "SHOW_PROGRESS": "False", + "PLUGINS": "__archivebox_test_no_plugins__", + }, + ) + return subprocess.run( + ["archivebox", "manage", "migrate", "core", target, "--noinput"], + cwd=data_dir, + env=env, + text=True, + capture_output=True, + timeout=60, + ) + + +def prepare_0026_database(tmp_path): + for dirname in ("archive", "sources", "logs"): + (tmp_path / dirname).mkdir() + + result = run_migration(tmp_path, MIGRATION_0026) + assert result.returncode == 0, result.stdout + result.stderr + + db_path = tmp_path / "index.sqlite3" + with sqlite3.connect(db_path) as db: + db.execute("PRAGMA foreign_keys = ON") + db.execute( + """ + INSERT INTO auth_user ( + id, password, is_superuser, username, first_name, last_name, + email, is_staff, is_active, date_joined + ) VALUES (?, '', 1, 'admin', '', '', '', 1, 1, ?) + """, + [USER_ID, NOW], + ) + db.execute( + """ + INSERT INTO crawls_crawl ( + id, created_at, modified_at, urls, created_by_id + ) VALUES (?, ?, ?, '[]', ?) + """, + [CRAWL_ID, NOW, NOW, USER_ID], + ) + db.execute( + """ + INSERT INTO core_snapshot ( + id, url, timestamp, bookmarked_at, created_at, modified_at, + fs_version, crawl_id, config, current_step, depth, notes, + num_uses_failed, num_uses_succeeded, status + ) VALUES (?, 'https://example.com', '20240101120000.000000', ?, ?, ?, + '0.8.5', ?, '{}', 0, 0, '', 0, 0, 'succeeded') + """, + [SNAPSHOT_ID, NOW, NOW, NOW, CRAWL_ID], + ) + + return db_path + + +def insert_archiveresult(db, *, result_id, command, pwd, version, status="succeeded"): + db.execute( + """ + INSERT INTO core_archiveresult ( + id, cmd, pwd, cmd_version, status, start_ts, end_ts, snapshot_id, + uuid, created_at, modified_at, config, hook_name, notes, + output_files, output_mimetypes, output_size, output_str, plugin + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '{}', '', '', '{}', '', 0, '', 'wget') + """, + [ + result_id, + json.dumps(command), + pwd, + version, + status, + NOW, + NOW, + SNAPSHOT_ID, + f"{result_id:032d}", + NOW, + NOW, + ], + ) + + +def table_columns(db, table): + return {row[1] for row in db.execute(f"PRAGMA table_info({table})")} + + +def test_process_migration_rolls_back_all_rows_and_preserves_legacy_metadata_on_error(tmp_path): + db_path = prepare_0026_database(tmp_path) + with sqlite3.connect(db_path) as db: + insert_archiveresult( + db, + result_id=1, + command=["/usr/bin/wget", "--page-requisites", "https://example.com"], + pwd="/data/archive/valid", + version="1.21.4", + ) + insert_archiveresult( + db, + result_id=2, + command=["/broken/wget", "https://example.com"], + pwd="/data/archive/malformed", + version="broken-version", + status="failed", + ) + db.execute( + """ + CREATE TRIGGER reject_malformed_process + BEFORE INSERT ON machine_process + WHEN NEW.cmd LIKE '%/broken/wget%' + BEGIN + SELECT RAISE(ABORT, 'malformed process metadata'); + END + """, + ) + + result = run_migration(tmp_path, MIGRATION_0027) + assert result.returncode != 0, result.stdout + result.stderr + assert "malformed process metadata" in result.stdout + result.stderr + + with sqlite3.connect(db_path) as db: + assert {"cmd", "cmd_version", "pwd"} <= table_columns(db, "core_archiveresult") + rows = db.execute( + "SELECT id, cmd, pwd, cmd_version, process_id FROM core_archiveresult ORDER BY id", + ).fetchall() + assert rows == [ + ( + 1, + json.dumps(["/usr/bin/wget", "--page-requisites", "https://example.com"]), + "/data/archive/valid", + "1.21.4", + None, + ), + ( + 2, + json.dumps(["/broken/wget", "https://example.com"]), + "/data/archive/malformed", + "broken-version", + None, + ), + ] + assert db.execute("SELECT COUNT(*) FROM machine_process").fetchone()[0] == 0 + assert db.execute("SELECT COUNT(*) FROM machine_binary").fetchone()[0] == 0 + assert ( + db.execute( + "SELECT COUNT(*) FROM django_migrations WHERE app = 'core' AND name = ?", + [MIGRATION_0027], + ).fetchone()[0] + == 0 + ) + + +def test_process_migration_links_every_row_and_preserves_execution_metadata(tmp_path): + db_path = prepare_0026_database(tmp_path) + command = ["/usr/bin/wget", "--page-requisites", "https://example.com"] + with sqlite3.connect(db_path) as db: + insert_archiveresult( + db, + result_id=1, + command=command, + pwd="/data/archive/valid", + version="1.21.4", + ) + + result = run_migration(tmp_path, MIGRATION_0027) + assert result.returncode == 0, result.stdout + result.stderr + + with sqlite3.connect(db_path) as db: + assert not ({"cmd", "cmd_version", "pwd"} & table_columns(db, "core_archiveresult")) + row = db.execute( + """ + SELECT process.cmd, process.pwd, binary.version, process.status, + process.exit_code, result.process_id + FROM core_archiveresult AS result + JOIN machine_process AS process ON process.id = result.process_id + JOIN machine_binary AS binary ON binary.id = process.binary_id + """, + ).fetchone() + assert row == (json.dumps(command), "/data/archive/valid", "1.21.4", "exited", 0, row[-1]) + assert row[-1] + assert db.execute("SELECT COUNT(*) FROM core_archiveresult WHERE process_id IS NULL").fetchone()[0] == 0