The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-14 11:06:13 +05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Make Process metadata migration transactional

Browse Source
This commit is contained in:
Nick Sweeting 2026-08-01 13:31:52 -07:00
parent bd338e23af
commit 06fbed2be2
No known key found for this signature in database
2 changed files with 267 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
archivebox/core/migrations/0027_copy_archiveresult_to_process.py
View File

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

199
archivebox/tests/test_migration_process_integrity.py Normal file
View File

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