From 804be3075f65ec0451b8bbd25947229d650d131f Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Thu, 28 May 2026 07:41:56 -0700 Subject: [PATCH] chore: checkpoint deploy loop changes --- archivebox/cli/archivebox_add.py | 26 ++++++++--- archivebox/cli/archivebox_update.py | 11 +++-- .../core/migrations/0023_upgrade_to_0_9_0.py | 46 +++++++++++++------ ...options_alter_snapshot_options_and_more.py | 7 +-- .../0027_copy_archiveresult_to_process.py | 33 ++++--------- .../0029_migrate_archiveresult_to_uuid_pk.py | 22 +++++---- .../migrations/0005_converge_binary_model.py | 11 ++--- 7 files changed, 87 insertions(+), 69 deletions(-) diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py index 4f360d49..950f3061 100644 --- a/archivebox/cli/archivebox_add.py +++ b/archivebox/cli/archivebox_add.py @@ -218,18 +218,30 @@ def add( # Foreground mode: run full crawl runner until all work is done print("[green]\\[*] Starting crawl runner to process crawl...[/green]") from archivebox.machine.models import Process - from archivebox.services.supervision_service import current_command, standby_until_runtime_stack_needed + from archivebox.services.supervision_service import command_owns_runtime_stack, current_command, standby_until_runtime_stack_needed from archivebox.workers.supervisord_util import run_runner_worker, stop_own_supervisord_process command = current_command(Process.TypeChoices.ADD, data_dir=CONSTANTS.DATA_DIR, url=first_url) + exit_code = 0 try: - standby_until_runtime_stack_needed(command, data_dir=CONSTANTS.DATA_DIR) - with foreground_shutdown_signals(), foreground_parent_watchdog(): - exit_code = run_runner_worker(["--crawl-id", str(crawl.id)], name=f"worker_runner_add_{os.getpid()}") - if exit_code != 0: - raise SystemExit(exit_code) + try: + with foreground_shutdown_signals(), foreground_parent_watchdog(): + while True: + standby_until_runtime_stack_needed(command, data_dir=CONSTANTS.DATA_DIR) + exit_code = run_runner_worker(["--crawl-id", str(crawl.id)], name=f"worker_runner_add_{os.getpid()}") + if exit_code == 0: + break + if not command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR): + continue + raise SystemExit(exit_code) + except KeyboardInterrupt: + exit_code = 130 + print("\n[red][X] archivebox add interrupted.[/red]") + print("[yellow]Hint: resume this crawl with:[/yellow]") + print(f" [green]archivebox run --crawl-id={crawl.id}[/green]") + raise SystemExit(exit_code) finally: - command.mark_exited() + command.mark_exited(exit_code=exit_code) stop_own_supervisord_process() # Print summary for foreground runs diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index f7c75bd7..710a2263 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -217,6 +217,7 @@ def update( from archivebox.machine.models import Process from archivebox.core.shutdown_util import foreground_parent_watchdog, foreground_shutdown_signals from archivebox.services.supervision_service import ( + command_owns_runtime_stack, current_command, ensure_daemon_stack, standby_until_runtime_stack_needed, @@ -229,9 +230,13 @@ def update( standby_until_runtime_stack_needed(command, data_dir=CONSTANTS.DATA_DIR) def run_scoped_runner(*args: str) -> None: - wait_for_turn() - exit_code = run_runner_worker(list(args), name=f"worker_runner_update_{os.getpid()}") - if exit_code != 0: + while True: + wait_for_turn() + exit_code = run_runner_worker(list(args), name=f"worker_runner_update_{os.getpid()}") + if exit_code == 0: + return + if not command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR): + continue raise SystemExit(exit_code) is_filtered_update = any( diff --git a/archivebox/core/migrations/0023_upgrade_to_0_9_0.py b/archivebox/core/migrations/0023_upgrade_to_0_9_0.py index d823625b..1dfb13ed 100644 --- a/archivebox/core/migrations/0023_upgrade_to_0_9_0.py +++ b/archivebox/core/migrations/0023_upgrade_to_0_9_0.py @@ -9,6 +9,9 @@ from datetime import datetime import json +PROGRESS_EVERY = 10000 + + def get_table_columns(table_name): """Get list of column names for a table.""" cursor = connection.cursor() @@ -64,7 +67,9 @@ def upgrade_core_tables(apps, schema_editor): has_uuid = "uuid" in archiveresult_cols has_abid = "abid" in archiveresult_cols - print(f"DEBUG: ArchiveResult row_count={row_count}, has_data={has_data}, has_uuid={has_uuid}, has_abid={has_abid}") + if has_data: + source_schema = "0.8.x abid" if has_abid and not has_uuid else "0.8.x uuid" if has_uuid else "0.7.x" + print(f" - Rebuilding core tables from {source_schema} schema ({row_count} ArchiveResults)...") # ============================================================================ # PART 1: Upgrade core_archiveresult table @@ -113,7 +118,7 @@ def upgrade_core_tables(apps, schema_editor): if has_uuid and not has_abid: # Migrating from v0.7.2+ (has uuid column) - print("Migrating ArchiveResult from v0.7.2+ schema (with uuid)...") + print(f" copying {row_count} ArchiveResults...") select_cols = ["id", "uuid", "snapshot_id", "cmd", "pwd", "cmd_version", "start_ts", "end_ts", "status", "extractor", "output"] if has_archiveresult_created_at: select_cols.append("created_at") @@ -121,7 +126,7 @@ def upgrade_core_tables(apps, schema_editor): select_cols.append("modified_at") cursor.execute(f"SELECT {', '.join(select_cols)} FROM core_archiveresult") old_records = cursor.fetchall() - for record in old_records: + for i, record in enumerate(old_records, start=1): values = dict(zip(select_cols, record)) try: new_uuid = UUID(str(values["uuid"])).hex @@ -153,12 +158,14 @@ def upgrade_core_tables(apps, schema_editor): values.get("modified_at") or end_ts, ), ) + if i % PROGRESS_EVERY == 0: + print(f" copied {i}/{len(old_records)} ArchiveResults...") elif has_abid and not has_uuid: # Migrating from v0.8.6rc0 (has abid instead of uuid) - print("Migrating ArchiveResult from v0.8.6rc0 schema...") + print(f" copying {row_count} ArchiveResults...") cursor.execute(f"SELECT {', '.join(archiveresult_select_cols)} FROM core_archiveresult") old_records = cursor.fetchall() - for record in old_records: + for i, record in enumerate(old_records, start=1): values = dict(zip(archiveresult_select_cols, record)) try: new_uuid = UUID(str(values["id"])).hex @@ -189,12 +196,14 @@ def upgrade_core_tables(apps, schema_editor): values.get("modified_at") or end_ts, ), ) + if i % PROGRESS_EVERY == 0: + print(f" copied {i}/{len(old_records)} ArchiveResults...") else: # Migrating from v0.7.2 (no uuid or abid column - generate fresh UUIDs) - print("Migrating ArchiveResult from v0.7.2 schema (no uuid - generating UUIDs)...") + print(f" copying {row_count} ArchiveResults...") cursor.execute(f"SELECT {', '.join(archiveresult_select_cols)} FROM core_archiveresult") old_records = cursor.fetchall() - for record in old_records: + for i, record in enumerate(old_records, start=1): values = dict(zip(archiveresult_select_cols, record)) new_uuid = uuid7().hex start_ts = values["start_ts"] or datetime.now().isoformat() @@ -223,6 +232,9 @@ def upgrade_core_tables(apps, schema_editor): values.get("modified_at") or end_ts, ), ) + if i % PROGRESS_EVERY == 0: + print(f" copied {i}/{len(old_records)} ArchiveResults...") + print(f" copied {len(old_records)} ArchiveResults") cursor.execute("DROP TABLE IF EXISTS core_archiveresult;") cursor.execute("ALTER TABLE core_archiveresult_new RENAME TO core_archiveresult;") @@ -279,7 +291,7 @@ def upgrade_core_tables(apps, schema_editor): if has_added and not has_bookmarked_at: # Migrating from v0.7.2 (has added/updated fields) - print("Migrating Snapshot from v0.7.2 schema...") + print(" copying Snapshots from 0.7.x schema...") # timestamp is the legacy bookmark/import timestamp and archive/{timestamp} identity. # added is the DB row creation/import time, and updated was renamed to downloaded_at in 0.8.x. cursor.execute(""" @@ -317,9 +329,10 @@ def upgrade_core_tables(apps, schema_editor): END as status FROM core_snapshot; """) + print(f" copied {cursor.rowcount} Snapshots") elif has_bookmarked_at and not has_added: # Migrating from v0.8.6rc0 (already has bookmarked_at/created_at/modified_at) - print("Migrating Snapshot from v0.8.6rc0 schema...") + print(" copying Snapshots from 0.8.x schema...") # Check what fields exist has_status = "status" in snapshot_cols has_retry_at = "retry_at" in snapshot_cols @@ -361,6 +374,7 @@ def upgrade_core_tables(apps, schema_editor): SELECT {", ".join(select_cols)} FROM core_snapshot; """) + print(f" copied {cursor.rowcount} Snapshots") else: print(f"Warning: Unexpected Snapshot schema - has_added={has_added}, has_bookmarked_at={has_bookmarked_at}") @@ -411,7 +425,7 @@ def upgrade_core_tables(apps, schema_editor): if tag_id_type and "char" in tag_id_type.lower(): # v0.8.6rc0: Tag IDs are UUIDs, need to convert to INTEGER - print("Converting Tag IDs from UUID to INTEGER...") + print(" converting Tag IDs from UUID to integers...") # Get all tags with their UUIDs cursor.execute("SELECT id, name, slug, created_at, modified_at, created_by_id FROM core_tag ORDER BY name") @@ -430,6 +444,8 @@ def upgrade_core_tables(apps, schema_editor): """, (i, name, slug, created_at, modified_at, created_by_id), ) + if i % PROGRESS_EVERY == 0: + print(f" copied {i}/{len(tags)} Tags...") # Update snapshot_tags to use new INTEGER IDs cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='core_snapshot_tags'") @@ -441,7 +457,7 @@ def upgrade_core_tables(apps, schema_editor): cursor.execute("DELETE FROM core_snapshot_tags") # Re-insert with new integer tag IDs - for st_id, snapshot_id, old_tag_id in snapshot_tags: + for i, (st_id, snapshot_id, old_tag_id) in enumerate(snapshot_tags, start=1): new_tag_id = uuid_to_int_map.get(old_tag_id) if new_tag_id: cursor.execute( @@ -451,14 +467,18 @@ def upgrade_core_tables(apps, schema_editor): """, (st_id, snapshot_id, new_tag_id), ) + if i % PROGRESS_EVERY == 0: + print(f" copied {i}/{len(snapshot_tags)} SnapshotTag rows...") + print(f" copied {len(tags)} Tags") else: # v0.7.2: Tag IDs are already INTEGER - print("Migrating Tag from v0.7.2 schema...") + print(" copying Tags from 0.7.x schema...") cursor.execute(""" INSERT OR IGNORE INTO core_tag_new (id, name, slug) SELECT id, name, slug FROM core_tag; """) + print(f" copied {cursor.rowcount} Tags") cursor.execute("DROP TABLE IF EXISTS core_tag;") cursor.execute("ALTER TABLE core_tag_new RENAME TO core_tag;") @@ -468,7 +488,7 @@ def upgrade_core_tables(apps, schema_editor): cursor.execute("CREATE INDEX IF NOT EXISTS core_tag_created_by_id_idx ON core_tag(created_by_id);") if has_data: - print("✓ Core tables upgraded to v0.9.0") + print(" ✓ Core table rebuild complete") class Migration(migrations.Migration): diff --git a/archivebox/core/migrations/0025_alter_archiveresult_options_alter_snapshot_options_and_more.py b/archivebox/core/migrations/0025_alter_archiveresult_options_alter_snapshot_options_and_more.py index 01e3e1a2..c4eb2ee8 100644 --- a/archivebox/core/migrations/0025_alter_archiveresult_options_alter_snapshot_options_and_more.py +++ b/archivebox/core/migrations/0025_alter_archiveresult_options_alter_snapshot_options_and_more.py @@ -40,12 +40,7 @@ def copy_old_fields_to_new(apps, schema_editor): # NOTE: Snapshot timestamps (added→bookmarked_at, updated→modified_at) were already # transformed by migration 0023, so we don't need to copy them here. - # NOTE: UUIDs are already populated by migration 0023 for all migration paths - - # Debug: Check Snapshot timestamps at end of RunPython - cursor.execute("SELECT id, bookmarked_at, modified_at FROM core_snapshot LIMIT 2") - snap_after = cursor.fetchall() - print(f"DEBUG 0025: Snapshot timestamps at END of RunPython: {snap_after}") + # NOTE: UUIDs are already populated by migration 0023 for all migration paths. class Migration(migrations.Migration): diff --git a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py index cc51e5cb..0b983425 100644 --- a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py +++ b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py @@ -7,6 +7,9 @@ from pathlib import Path from archivebox.uuid_compat import uuid7 +PROGRESS_EVERY = 10000 + + def parse_cmd_field(cmd_raw): """ Parse cmd field which could be: @@ -278,13 +281,8 @@ def copy_archiveresult_data_to_process(apps, schema_editor): cursor.execute("PRAGMA table_info(core_archiveresult)") cols = {row[1] for row in cursor.fetchall()} - print(f"DEBUG 0027: Columns found: {sorted(cols)}") - print( - f"DEBUG 0027: Has cmd={('cmd' in cols)}, pwd={('pwd' in cols)}, cmd_version={('cmd_version' in cols)}, process_id={('process_id' in cols)}", - ) - if "cmd" not in cols or "pwd" not in cols or "cmd_version" not in cols: - print("✓ Fresh install or fields already removed - skipping data copy") + print(" ✓ ArchiveResult process metadata already migrated") return # Check if process_id field exists (should exist from 0026) @@ -308,10 +306,10 @@ def copy_archiveresult_data_to_process(apps, schema_editor): results = cursor.fetchall() if not results: - print("✓ No ArchiveResults need Process migration") + print(" ✓ No ArchiveResults need Process migration") return - print(f"Migrating {len(results)} ArchiveResults to Process records...") + print(f" - Migrating {len(results)} ArchiveResults to Process rows...") migrated_count = 0 skipped_count = 0 @@ -320,16 +318,10 @@ def copy_archiveresult_data_to_process(apps, schema_editor): for i, row in enumerate(results): ar_id, snapshot_id, plugin, cmd_raw, pwd, cmd_version, status, start_ts, end_ts, created_at = row - if i == 0: - print(f"DEBUG 0027: First row: ar_id={ar_id}, plugin={plugin}, cmd={cmd_raw[:50] if cmd_raw else None}, status={status}") - try: # Parse cmd field cmd_array = parse_cmd_field(cmd_raw) - if i == 0: - print(f"DEBUG 0027: Parsed cmd: {cmd_array}") - # Extract binary info from cmd[0] if available binary_id = None if cmd_array and cmd_array[0]: @@ -346,9 +338,6 @@ def copy_archiveresult_data_to_process(apps, schema_editor): binary_version, ) - if i == 0: - print(f"DEBUG 0027: Created Binary: id={binary_id}, name={binary_name}") - # Map status process_status, exit_code = map_status(status) @@ -369,9 +358,6 @@ def copy_archiveresult_data_to_process(apps, schema_editor): binary_id=binary_id, ) - if i == 0: - print(f"DEBUG 0027: Created Process: id={process_id}") - # Link ArchiveResult to Process cursor.execute( "UPDATE core_archiveresult SET process_id = ? WHERE id = ?", @@ -379,9 +365,8 @@ def copy_archiveresult_data_to_process(apps, schema_editor): ) migrated_count += 1 - - if i == 0: - print("DEBUG 0027: Linked ArchiveResult to Process") + if migrated_count % PROGRESS_EVERY == 0: + print(f" migrated {migrated_count}/{len(results)} ArchiveResults...") except Exception as e: print(f"✗ Error migrating ArchiveResult {ar_id}: {e}") @@ -391,7 +376,7 @@ def copy_archiveresult_data_to_process(apps, schema_editor): error_count += 1 continue - print(f"✓ Migration complete: {migrated_count} migrated, {skipped_count} skipped, {error_count} errors") + print(f" ✓ Process migration complete: {migrated_count} migrated, {skipped_count} skipped, {error_count} errors") class Migration(migrations.Migration): diff --git a/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py b/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py index 7ed7d36e..64045aad 100644 --- a/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py +++ b/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py @@ -6,6 +6,9 @@ from uuid import UUID from archivebox.uuid_compat import uuid7 +PROGRESS_EVERY = 10000 + + def migrate_archiveresult_id_to_uuid(apps, schema_editor): """ Migrate ArchiveResult from integer PK to UUID PK (clean one-step migration). @@ -28,7 +31,7 @@ def migrate_archiveresult_id_to_uuid(apps, schema_editor): # Check if table exists and has data cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='core_archiveresult'") if not cursor.fetchone(): - print("ArchiveResult table does not exist, skipping migration") + print(" ✓ ArchiveResult table does not exist, skipping UUID PK migration") return cursor.execute("SELECT COUNT(*) FROM core_archiveresult") @@ -38,16 +41,16 @@ def migrate_archiveresult_id_to_uuid(apps, schema_editor): # (fresh installs create table with uuid from 0025, but model expects no uuid after 0029) if row_count == 0: - print("[0029] Recreating ArchiveResult table schema (integer→UUID PK, removing uuid column)...") + print(" - Rebuilding empty ArchiveResult table with UUID primary keys...") else: - print(f"[0029] Migrating {row_count} ArchiveResult records from integer PK to UUID PK...") + print(f" - Migrating {row_count} ArchiveResults from integer IDs to UUID primary keys...") # Step 0: Check if machine_process table exists, if not NULL out process_id values cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='machine_process'") machine_process_exists = cursor.fetchone() is not None if not machine_process_exists: - print("machine_process table does not exist yet, setting process_id to NULL") + print(" machine_process is unavailable; clearing process_id references...") cursor.execute("UPDATE core_archiveresult SET process_id = NULL WHERE process_id IS NOT NULL") # Step 1: Create new table with UUID as primary key (clean - no old_id or uuid columns) @@ -149,10 +152,6 @@ def migrate_archiveresult_id_to_uuid(apps, schema_editor): # Build INSERT statement (only copy fields that exist in source) existing_fields = [f for f in fields_to_copy if f in values] - if i == 0: - print(f"[0029] Source columns: {col_names}") - print(f"[0029] Copying fields: {existing_fields}") - placeholders = ", ".join(["?"] * (len(existing_fields) + 1)) # +1 for id field_list = "id, " + ", ".join(existing_fields) @@ -164,13 +163,16 @@ def migrate_archiveresult_id_to_uuid(apps, schema_editor): insert_values, ) inserted_count += 1 + if inserted_count % PROGRESS_EVERY == 0: + print(f" copied {inserted_count}/{len(old_records)} ArchiveResults...") except Exception as e: print(f"[0029] ERROR inserting record {old_id}: {e}") if i == 0: print(f"[0029] First record values: {insert_values[:5]}...") raise - print(f"[0029] Inserted {inserted_count}/{len(old_records)} records") + if old_records: + print(f" copied {inserted_count}/{len(old_records)} ArchiveResults") # Step 4: Replace old table with new table cursor.execute("DROP TABLE core_archiveresult") @@ -185,7 +187,7 @@ def migrate_archiveresult_id_to_uuid(apps, schema_editor): cursor.execute("CREATE INDEX core_archiveresult_hook_name_idx ON core_archiveresult(hook_name)") cursor.execute("CREATE INDEX core_archiveresult_process_id_idx ON core_archiveresult(process_id)") - print(f"✓ Migrated {row_count} ArchiveResult records to UUID primary key") + print(f" ✓ ArchiveResult UUID primary key migration complete ({row_count} records)") class Migration(migrations.Migration): diff --git a/archivebox/machine/migrations/0005_converge_binary_model.py b/archivebox/machine/migrations/0005_converge_binary_model.py index 7e861968..c1e89175 100644 --- a/archivebox/machine/migrations/0005_converge_binary_model.py +++ b/archivebox/machine/migrations/0005_converge_binary_model.py @@ -16,17 +16,15 @@ def converge_binary_table(apps, schema_editor): cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('machine_installedbinary', 'machine_binary')") existing_tables = {row[0] for row in cursor.fetchall()} - print(f"DEBUG 0005: Existing tables: {existing_tables}") - # Drop old Binary table if it exists (0.8.6rc0 path) if "machine_installedbinary" in existing_tables: - print("✓ Dropping machine_installedbinary table (0.8.6rc0 divergence)") + print(" - Removing old machine_installedbinary table...") cursor.execute("DROP TABLE IF EXISTS machine_installedbinary") # Create Binary table if it doesn't exist. # This handles the case where 0.8.6rc0's 0001_initial didn't create it. if "machine_binary" not in existing_tables: - print("✓ Creating machine_binary table with correct schema") + print(" - Creating machine_binary table...") cursor.execute(""" CREATE TABLE machine_binary ( id TEXT PRIMARY KEY NOT NULL, @@ -53,9 +51,9 @@ def converge_binary_table(apps, schema_editor): cursor.execute("CREATE INDEX machine_binary_name_idx ON machine_binary(name)") cursor.execute("CREATE INDEX machine_binary_abspath_idx ON machine_binary(abspath)") - print("✓ machine_binary table created") + print(" ✓ machine_binary table ready") else: - print("✓ machine_binary table already exists") + print(" - Converging existing machine_binary table...") cursor.execute("PRAGMA table_info(machine_binary)") binary_cols = {row[1] for row in cursor.fetchall()} @@ -78,6 +76,7 @@ def converge_binary_table(apps, schema_editor): ) cursor.execute("UPDATE machine_binary SET overrides = COALESCE(NULLIF(overrides, ''), '{}')") cursor.execute("UPDATE machine_binary SET status = COALESCE(NULLIF(status, ''), 'installed')") + print(" ✓ machine_binary table ready") class Migration(migrations.Migration):