The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-12 19:50:57 +05:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

chore: checkpoint dev stack deploy changes

Browse Source
This commit is contained in:
Nick Sweeting 2026-05-28 06:39:06 -07:00
parent b1fa3de5a5
commit b2fa839788
No known key found for this signature in database
10 changed files with 268 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
archivebox/cli/archivebox_add.py
View File

@ -111,6 +111,7 @@ def add(
from archivebox.personas.models import Persona
from archivebox.misc.logging_util import printable_filesize
from archivebox.misc.system import get_dir_size
from archivebox.core.shutdown_util import foreground_parent_watchdog
from archivebox.services.runner import run_crawl
from django.utils import timezone
@ -216,7 +217,8 @@ def add(
else:
# Foreground mode: run full crawl runner until all work is done
print("[green]\\[*] Starting crawl runner to process crawl...[/green]")
run_crawl(str(crawl.id))
with foreground_parent_watchdog():
run_crawl(str(crawl.id))
# Print summary for foreground runs
try:

38
archivebox/cli/archivebox_run.py
View File

@ -77,6 +77,7 @@ def process_stdin_records() -> int:
from archivebox.base_models.models import get_or_create_system_user_pk
from archivebox.core.models import Snapshot, ArchiveResult
from archivebox.crawls.models import Crawl
from archivebox.core.shutdown_util import foreground_parent_watchdog
from archivebox.machine.models import Binary
from archivebox.services.runner import run_binary, run_crawl
@ -228,11 +229,12 @@ def process_stdin_records() -> int:
if not crawl.claim_processing_lock(lock_seconds=10):
rprint(f"[yellow]Crawl {crawl_id} is already owned by another runner[/yellow]", file=sys.stderr)
return 1
run_crawl(
crawl_id,
snapshot_ids=None if crawl_id in full_crawl_ids else sorted(snapshot_ids_by_crawl[crawl_id]),
selected_plugins=None if crawl_id in run_all_plugins_for_crawl else sorted(plugin_names_by_crawl[crawl_id]),
)
with foreground_parent_watchdog():
run_crawl(
crawl_id,
snapshot_ids=None if crawl_id in full_crawl_ids else sorted(snapshot_ids_by_crawl[crawl_id]),
selected_plugins=None if crawl_id in run_all_plugins_for_crawl else sorted(plugin_names_by_crawl[crawl_id]),
)
return 0
@ -246,6 +248,7 @@ def run_runner(daemon: bool = False, crawl_id: str | None = None) -> int:
Returns exit code (0 = success, 1 = error).
"""
from archivebox.config import CONSTANTS
from archivebox.core.shutdown_util import foreground_parent_watchdog
from archivebox.machine.models import Machine, Process
from archivebox.services.supervision_service import healthy_orchestrator
from archivebox.services.runner import recover_orchestrator_state, run_pending_crawls
@ -260,7 +263,8 @@ def run_runner(daemon: bool = False, crawl_id: str | None = None) -> int:
return 0
current.mark_running(process_type=Process.TypeChoices.ORCHESTRATOR, pwd=str(CONSTANTS.DATA_DIR), timeout=0)
try:
run_pending_crawls(daemon=daemon, crawl_id=crawl_id)
with foreground_parent_watchdog(enabled=not daemon):
run_pending_crawls(daemon=daemon, crawl_id=crawl_id)
return 0
except KeyboardInterrupt:
return 0
@ -320,18 +324,30 @@ def main(daemon: bool, crawl_id: str, snapshot_id: str, binary_id: str):
def run_snapshot_worker(snapshot_id: str) -> int:
from archivebox.core.shutdown_util import foreground_parent_watchdog
from archivebox.core.models import Snapshot
from archivebox.services.runner import run_due_snapshot
from django.utils import timezone
snapshot = None
try:
snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id)
if snapshot.retry_at is None:
Snapshot.objects.filter(pk=snapshot.pk).update(retry_at=timezone.now(), modified_at=timezone.now())
snapshot.refresh_from_db()
run_due_snapshot(snapshot, lock_seconds=60)
with foreground_parent_watchdog():
snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id)
if snapshot.retry_at is None:
Snapshot.objects.filter(pk=snapshot.pk).update(retry_at=timezone.now(), modified_at=timezone.now())
snapshot.refresh_from_db()
run_due_snapshot(snapshot, lock_seconds=60)
return 0
except KeyboardInterrupt:
try:
if snapshot is not None:
snapshot.refresh_from_db()
else:
snapshot = Snapshot.objects.filter(id=snapshot_id).first()
if snapshot is not None and snapshot.status != Snapshot.StatusChoices.SEALED:
snapshot.update_and_requeue(retry_at=timezone.now())
except Exception:
pass
return 0
except Exception as e:
rprint(f"[red]Runner error: {type(e).__name__}: {e}[/red]", file=sys.stderr)

190
archivebox/cli/archivebox_update.py
View File

@ -201,6 +201,7 @@ def update(
setup_django()
from archivebox.machine.models import Process
from archivebox.core.shutdown_util import foreground_parent_watchdog
from archivebox.services.supervision_service import current_command, ensure_daemon_stack
from archivebox.workers.supervisord_util import stop_existing_supervisord_process
@ -231,110 +232,111 @@ def update(
if stop_daemon_stack:
stop_existing_supervisord_process()
while True:
do_migrate = migrate_only or not index_only
do_index = index_only or not migrate_only
do_run_until_idle = do_migrate or do_index
with foreground_parent_watchdog():
while True:
do_migrate = migrate_only or not index_only
do_index = index_only or not migrate_only
do_run_until_idle = do_migrate or do_index
if do_migrate:
if (
filter_patterns
or status
or url__icontains
or url__istartswith
or tag
or crawl_id
or limit
or sort
or search
or before
or after
):
print("[*] Processing filtered snapshots from database...")
stats = process_filtered_snapshots(
filter_patterns=filter_patterns,
filter_type=filter_type,
status=status,
url__icontains=url__icontains,
url__istartswith=url__istartswith,
tag=tag,
crawl_id=crawl_id,
limit=limit,
sort=sort,
search=search,
before=before,
after=after,
resume=resume,
batch_size=batch_size,
queue_for_archiving=do_run_until_idle,
)
print_stats(stats)
touched_snapshot_ids.update(stats.get("snapshot_ids", []))
else:
stats_combined = {"phase1": {}, "phase2": {}}
if do_migrate:
if (
filter_patterns
or status
or url__icontains
or url__istartswith
or tag
or crawl_id
or limit
or sort
or search
or before
or after
):
print("[*] Processing filtered snapshots from database...")
stats = process_filtered_snapshots(
filter_patterns=filter_patterns,
filter_type=filter_type,
status=status,
url__icontains=url__icontains,
url__istartswith=url__istartswith,
tag=tag,
crawl_id=crawl_id,
limit=limit,
sort=sort,
search=search,
before=before,
after=after,
resume=resume,
batch_size=batch_size,
queue_for_archiving=do_run_until_idle,
)
print_stats(stats)
touched_snapshot_ids.update(stats.get("snapshot_ids", []))
else:
stats_combined = {"phase1": {}, "phase2": {}}
print("[*] Phase 1: Draining old archive/ directories (0.8.x → 0.9.x migration)...")
stats_combined["phase1"] = drain_old_archive_dirs(
resume_from=resume,
batch_size=batch_size,
)
print("[*] Phase 1: Draining old archive/ directories (0.8.x → 0.9.x migration)...")
stats_combined["phase1"] = drain_old_archive_dirs(
resume_from=resume,
batch_size=batch_size,
)
print("[*] Phase 2: Processing all database snapshots (most recent first)...")
stats_combined["phase2"] = process_all_db_snapshots(batch_size=batch_size, resume=resume)
print_combined_stats(stats_combined)
print("[*] Phase 2: Processing all database snapshots (most recent first)...")
stats_combined["phase2"] = process_all_db_snapshots(batch_size=batch_size, resume=resume)
print_combined_stats(stats_combined)
if do_index:
ensure_daemon_stack(reason="search indexing")
search_plugins = _get_search_indexing_plugins()
if not search_plugins:
print("[*] No search indexing plugins are available, nothing to backfill.")
else:
snapshots = _build_filtered_snapshots_queryset(
filter_patterns=filter_patterns,
filter_type=filter_type,
status=status,
url__icontains=url__icontains,
url__istartswith=url__istartswith,
tag=tag,
crawl_id=crawl_id,
limit=limit,
sort=sort,
search=search,
before=before,
after=after,
resume=resume,
)
stats = reindex_snapshots(
snapshots,
search_plugins=search_plugins,
batch_size=batch_size,
collect_ids=is_filtered_update,
)
print_index_stats(stats)
touched_snapshot_ids.update(stats.get("snapshot_ids", []))
if do_index:
ensure_daemon_stack(reason="search indexing")
search_plugins = _get_search_indexing_plugins()
if not search_plugins:
print("[*] No search indexing plugins are available, nothing to backfill.")
else:
snapshots = _build_filtered_snapshots_queryset(
filter_patterns=filter_patterns,
filter_type=filter_type,
status=status,
url__icontains=url__icontains,
url__istartswith=url__istartswith,
tag=tag,
crawl_id=crawl_id,
limit=limit,
sort=sort,
search=search,
before=before,
after=after,
resume=resume,
)
stats = reindex_snapshots(
snapshots,
search_plugins=search_plugins,
batch_size=batch_size,
collect_ids=is_filtered_update,
)
print_index_stats(stats)
touched_snapshot_ids.update(stats.get("snapshot_ids", []))
if do_run_until_idle:
print("[*] Phase 3: Running queued/interrupted crawl work until idle...")
from archivebox.cli.archivebox_run import run_runner, run_snapshot_worker
if do_run_until_idle:
print("[*] Phase 3: Running queued/interrupted crawl work until idle...")
from archivebox.cli.archivebox_run import run_runner, run_snapshot_worker
if is_filtered_update:
if not touched_snapshot_ids:
print("[*] No matching snapshots queued work for the runner.")
for snapshot_id in sorted(touched_snapshot_ids):
exit_code = run_snapshot_worker(snapshot_id)
if is_filtered_update:
if not touched_snapshot_ids:
print("[*] No matching snapshots queued work for the runner.")
for snapshot_id in sorted(touched_snapshot_ids):
exit_code = run_snapshot_worker(snapshot_id)
if exit_code != 0:
raise SystemExit(exit_code)
else:
exit_code = run_runner(daemon=False)
if exit_code != 0:
raise SystemExit(exit_code)
else:
exit_code = run_runner(daemon=False)
if exit_code != 0:
raise SystemExit(exit_code)
if not continuous:
break
if not continuous:
break
print("[yellow]Sleeping 60s before next pass...[/yellow]")
time.sleep(60)
resume = None
print("[yellow]Sleeping 60s before next pass...[/yellow]")
time.sleep(60)
resume = None
except (KeyboardInterrupt, asyncio.CancelledError) as err:
exact_resume = getattr(err, "archivebox_resume", None)
resume_cmd = ["archivebox", "update"]

43
archivebox/core/shutdown_util.py
View File

@ -3,6 +3,8 @@ from __future__ import annotations
import signal
import subprocess
import sys
import os
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
@ -110,3 +112,44 @@ def foreground_shutdown_signals(
finally:
for sig, previous_handler in previous_handlers.items():
signal.signal(sig, previous_handler)
@contextmanager
def foreground_parent_watchdog(
*,
enabled: bool = True,
check_interval: float = 2.0,
shutdown_signal: signal.Signals = signal.SIGINT,
) -> Iterator[None]:
"""Ask a foreground command to exit if its launcher/wrapper disappears.
`uv run archivebox ...` and similar wrappers can be killed without
delivering a signal to the real Python child. If that child keeps crawling
as an orphan, it can hold SQLite write locks long after the user-facing
command timed out. This watchdog is only for foreground command lifetimes;
daemon/supervisord workers should not use it because their parent may
intentionally hand them off.
"""
original_ppid = os.getppid()
if not enabled or original_ppid <= 1:
yield
return
stopped = threading.Event()
def watch_parent() -> None:
while not stopped.wait(check_interval):
if os.getppid() == original_ppid:
continue
sys.stderr.write("\n[🛑] ArchiveBox parent process exited; stopping foreground command gracefully...\n")
sys.stderr.flush()
os.kill(os.getpid(), shutdown_signal)
return
thread = threading.Thread(target=watch_parent, name="archivebox-parent-watchdog", daemon=True)
thread.start()
try:
yield
finally:
stopped.set()

22
archivebox/search/sonic_daemon.py
View File

@ -1,6 +1,8 @@
from __future__ import annotations
import asyncio
import json
import time
from abx_dl.events import ProcessStdoutEvent
@ -21,14 +23,32 @@ def register_sonic_daemon_event_handler(bus) -> None:
from archivebox.workers.supervisord_util import get_existing_supervisord_process, get_worker
daemon_event = SonicDaemonStartEvent.from_record(record)
deadline = time.monotonic() + 30.0
last_worker = None
while time.monotonic() < deadline:
if is_port_listening(daemon_event.host, daemon_event.port):
return
supervisor = get_existing_supervisord_process(quiet=True)
worker = get_worker(supervisor, daemon_event.worker_name) if supervisor is not None else None
if isinstance(worker, dict):
last_worker = worker
if worker.get("statename") not in {"STARTING", "RUNNING", "BACKOFF", "STOPPING"}:
break
await asyncio.sleep(0.5)
if is_port_listening(daemon_event.host, daemon_event.port):
return
supervisor = get_existing_supervisord_process()
supervisor = get_existing_supervisord_process(quiet=True)
if supervisor is None:
raise RuntimeError("Sonic search backend is required, but ArchiveBox supervisord is not running")
worker = get_worker(supervisor, daemon_event.worker_name)
if worker is None and last_worker is not None:
worker = last_worker
if not worker:
raise RuntimeError(f"Sonic search backend worker is not configured: {daemon_event.worker_name}")
if worker.get("statename") != "RUNNING":

12
archivebox/templates/admin/progress_monitor.html
View File

@ -237,9 +237,9 @@
#progress-monitor .screencast-text {
min-width: 0;
}
#progress-monitor .screencast-title {
display: block;
color: #f0f6fc;
#progress-monitor .screencast-title {
display: block;
color: #f0f6fc;
font-size: 12px;
font-weight: 600;
line-height: 1.25;
@ -247,9 +247,9 @@
overflow: hidden;
text-overflow: ellipsis;
}
#progress-monitor .screencast-url {
display: block;
color: #8b949e;
#progress-monitor .screencast-url {
display: block;
color: #8b949e;
font-family: 'Monaco', 'Menlo', 'Ubuntu Mono', monospace;
font-size: 10px;
line-height: 1.25;

43
archivebox/tests/test_ui_admin_views.py
View File

@ -186,36 +186,38 @@ class TestSnapshotProgressStats:
stats = snapshot.get_progress_stats()
assert stats["is_sealed"] is True
def test_archive_size_uses_prefetched_results_without_output_dir(self, snapshot, monkeypatch):
"""archive_size should use prefetched ArchiveResult sizes before touching disk."""
def test_archive_size_uses_materialized_output_size_without_output_dir(self, snapshot, monkeypatch, django_capture_on_commit_callbacks):
"""archive_size should trust the materialized DB size without touching disk."""
from archivebox.core.models import ArchiveResult, Snapshot
ArchiveResult.objects.create(
snapshot=snapshot,
plugin="wget",
status="succeeded",
output_size=4096,
)
prefetched_snapshot = Snapshot.objects.prefetch_related("archiveresult_set").get(pk=snapshot.pk)
with django_capture_on_commit_callbacks(execute=True):
ArchiveResult.objects.create(
snapshot=snapshot,
plugin="wget",
status="succeeded",
output_size=4096,
)
snapshot.refresh_from_db(fields=["output_size"])
def _output_dir_should_not_be_used(self):
raise AssertionError("archive_size should not access Snapshot.output_dir when results are prefetched")
monkeypatch.setattr(Snapshot, "output_dir", property(_output_dir_should_not_be_used), raising=False)
assert prefetched_snapshot.archive_size == 4096
assert snapshot.archive_size == 4096
def test_snapshot_serialization_exposes_output_size_alias(self, snapshot):
def test_snapshot_serialization_exposes_output_size_alias(self, snapshot, django_capture_on_commit_callbacks):
"""Snapshot serializers should expose output_size as an alias of archive_size."""
from archivebox.core.models import ArchiveResult
ArchiveResult.objects.create(
snapshot=snapshot,
plugin="wget",
status="succeeded",
output_size=4096,
)
with django_capture_on_commit_callbacks(execute=True):
ArchiveResult.objects.create(
snapshot=snapshot,
plugin="wget",
status="succeeded",
output_size=4096,
)
snapshot.refresh_from_db(fields=["output_size"])
assert snapshot.to_dict()["archive_size"] == 4096
assert snapshot.to_dict()["output_size"] == 4096
@ -1262,10 +1264,10 @@ class TestArchiveResultAdminListView:
assert b"Snapshot Feed" in response.content
assert b"/api/v1/core/snapshots.rss?created_by=testadmin&amp;limit=50&amp;api_key=" in response.content
def test_archiveresult_model_has_no_retry_at_field(self):
def test_archiveresult_model_has_retry_at_field(self):
from archivebox.core.models import ArchiveResult
assert "retry_at" not in {field.name for field in ArchiveResult._meta.fields}
assert "retry_at" in {field.name for field in ArchiveResult._meta.fields}
class TestLiveProgressView:
@ -1803,6 +1805,7 @@ class TestPublicIndexSearch:
from archivebox.core.views import PublicIndexView
monkeypatch.setenv("SEARCH_BACKEND_ENGINE", "ripgrep")
monkeypatch.setenv("USE_INDEXING_BACKEND", "false")
metadata_snapshot = Snapshot.objects.create(
url="https://public-example.com/google-meta",
title="Google Metadata Match",

4
archivebox/workers/supervisord_util.py
View File

@ -501,12 +501,12 @@ def start_new_supervisord_process(daemonize=False):
return wait_for_supervisord_ready()
def wait_for_supervisord_ready(max_wait_sec: float = 5.0, interval_sec: float = 0.1):
def wait_for_supervisord_ready(max_wait_sec: float = 5.0, interval_sec: float = 0.1, *, quiet: bool = False):
"""Poll for supervisord readiness without a fixed startup sleep."""
deadline = time.monotonic() + max_wait_sec
supervisor = None
while time.monotonic() < deadline:
supervisor = get_existing_supervisord_process()
supervisor = get_existing_supervisord_process(quiet=quiet)
if supervisor is not None:
return supervisor
time.sleep(interval_sec)

42
bin/fuzz_test.sh
View File

@ -73,15 +73,31 @@ random_start_delay() {
random_between 0 "$SLEEP_BETWEEN_JOBS_MAX"
}
kill_tree() {
tree_pids() {
local pid="$1"
local child
if command -v pgrep >/dev/null 2>&1; then
for child in $(pgrep -P "$pid" 2>/dev/null || true); do
kill_tree "$child"
tree_pids "$child"
done
fi
echo "$pid"
}
kill_tree() {
local pid="$1"
local signal="${2:-TERM}"
local tree_file="${3:-}"
if [[ -n "$tree_file" ]]; then
tree_pids "$pid" >> "$tree_file"
sort -u "$tree_file" | while IFS= read -r tree_pid; do
kill "-$signal" "$tree_pid" >/dev/null 2>&1 || true
done
else
tree_pids "$pid" | while IFS= read -r tree_pid; do
kill "-$signal" "$tree_pid" >/dev/null 2>&1 || true
done
fi
kill "$pid" >/dev/null 2>&1 || true
}
cleanup() {
@ -102,11 +118,13 @@ run_with_timeout() {
local label="$1"
shift 1
local slug token timeout
local slug token timeout timeout_marker tree_file
timeout="$(random_kill_after)"
slug="$(echo "$label" | tr ' /:' '____' | tr -cd '[:alnum:]_.-')"
token="$(date +%s).$RANDOM.$RANDOM"
local logfile="$LOG_DIR/${slug}.${token}.log"
timeout_marker="$LOG_DIR/.timeout.${token}"
tree_file="$LOG_DIR/.tree.${token}"
{
echo "[$(ts)] START label=$label shell=$$ data=$DATA_DIR"
@ -123,9 +141,11 @@ run_with_timeout() {
sleep "$timeout"
if kill -0 "$child" >/dev/null 2>&1; then
echo "[$(ts)] TIMEOUT label=$label pid=$child after=${timeout}s" >> "$logfile"
kill "$child" >/dev/null 2>&1 || true
: > "$timeout_marker"
: > "$tree_file"
kill_tree "$child" TERM "$tree_file"
sleep 5
kill -9 "$child" >/dev/null 2>&1 || true
kill_tree "$child" KILL "$tree_file"
fi
) &
local watchdog=$!
@ -134,8 +154,14 @@ run_with_timeout() {
wait "$child"
local code=$?
kill "$watchdog" >/dev/null 2>&1 || true
if [[ -e "$timeout_marker" ]]; then
wait "$watchdog" >/dev/null 2>&1 || true
else
kill "$watchdog" >/dev/null 2>&1 || true
fi
wait "$watchdog" >/dev/null 2>&1 || true
kill_tree "$child"
rm -f "$timeout_marker" "$tree_file"
trap - INT TERM
echo "[$(ts)] END label=$label pid=$child exit=$code log=$logfile" | tee -a "$logfile"
@ -170,7 +196,7 @@ run_server_for_a_bit() {
echo "[$(ts)] STOP label=$label pid=$child after=${hold}s" | tee -a "$logfile"
kill_tree "$child"
sleep 5
kill -9 "$child" >/dev/null 2>&1 || true
kill_tree "$child" KILL
wait "$child" >/dev/null 2>&1
local code=$?
trap - INT TERM

13
bin/take_screenshot.js
View File

@ -22,6 +22,7 @@ Environment:
SCREENSHOT_HEIGHT Viewport height, defaults to 1400
SCREENSHOT_FULL_PAGE Set to 1 to capture the full page, defaults to viewport only
SCREENSHOT_SCROLL_SELECTOR Scroll this selector into view before capture
SCREENSHOT_WAIT_SELECTOR Wait for this selector before capture
SCREENSHOT_SNAPSHOT_VIEW Set to list or grid before loading the page
SCREENSHOT_RESET_FILTERS Set to 1 to clear the admin filter collapsed preference
`);
@ -94,6 +95,9 @@ async function main() {
await page.waitForSelector('body');
await page.waitForSelector('#progress-monitor, #add-form', { timeout: 5000 }).catch(() => {});
if (process.env.SCREENSHOT_WAIT_SELECTOR) {
await page.waitForSelector(process.env.SCREENSHOT_WAIT_SELECTOR, { timeout: 45000 });
}
if (process.env.SCREENSHOT_SCROLL_SELECTOR) {
await page.waitForSelector(process.env.SCREENSHOT_SCROLL_SELECTOR, { timeout: 45000 }).catch(() => {});
await page.evaluate((selector) => {
@ -120,6 +124,15 @@ async function main() {
: 'missing',
progressCrawls: document.querySelectorAll('#progress-monitor .crawl-item').length,
progressSnapshots: document.querySelectorAll('#progress-monitor .snapshot-item').length,
screencastVisible: Boolean(document.querySelector('#progress-monitor .screencast-panel.visible')),
screencastImageLoaded: (() => {
const img = document.querySelector('#progress-monitor .screencast-panel.visible img');
return Boolean(img && img.complete && img.naturalWidth > 0 && img.naturalHeight > 0);
})(),
screencastImageSize: (() => {
const img = document.querySelector('#progress-monitor .screencast-panel.visible img');
return img ? `${img.naturalWidth}x${img.naturalHeight}` : '';
})(),
snapshotEmbed: Boolean(document.querySelector('.crawl-snapshots-embed iframe')),
addForm: Boolean(document.querySelector('#add-form')),
limitFields: Array.from(document.querySelectorAll('.crawl-limit-field label')).map((el) => el.textContent.trim()),