bvandeusen
b6a917ac81
Operator hit 3 large PNGs stuck in 'processing' for 2 days 2026-05-25:
the existing recover_interrupted_tasks flips processing > 5min back to
queued + .delay(), but if the underlying file is unfixably broken (e.g.,
PIL OSError, also patched in 68cffce), the loop never terminates and the
'Scanning...' banner sticks at 0/0 forever blocking new scans.
/api/import/clear-stuck:
- Flips every task in pending/queued/processing to 'failed' with a clear
marker error message
- Finalizes any 'running' ImportBatch that has no remaining active children
- Idempotent + non-destructive: rows survive, can be retried once the
underlying cause is resolved
UI button 'Clear stuck...' sits next to 'Retry failed' / 'Clear completed'
with a warning-tonal alert in the confirm dialog explaining what it does
and recommending Retry failed once the cause is fixed.
Tests: clears mixed non-terminal states, untouches complete rows,
finalizes orphan batch, no-op when nothing stuck.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
220 lines
8.0 KiB
Python
220 lines
8.0 KiB
Python
"""Import admin API: trigger scan, list tasks, retry, clear."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
from sqlalchemy import delete, select, update
|
|
|
|
from ..extensions import get_session
|
|
from ..models import ImportBatch, ImportTask
|
|
|
|
import_admin_bp = Blueprint("import_admin", __name__, url_prefix="/api/import")
|
|
|
|
|
|
@import_admin_bp.route("/trigger", methods=["POST"])
|
|
async def trigger_scan():
|
|
body = await request.get_json(silent=True) or {}
|
|
mode = body.get("mode", "quick")
|
|
if mode not in ("quick", "deep", "verify"):
|
|
return jsonify({"error": f"mode {mode!r} not supported; use 'quick', 'deep', or 'verify'"}), 400
|
|
|
|
# 'verify' is a library task — short-circuit the import_root walk
|
|
# (no ImportBatch, no per-file ImportTasks).
|
|
if mode == "verify":
|
|
from ..tasks.maintenance import verify_integrity
|
|
|
|
async_result = verify_integrity.delay()
|
|
return jsonify({"celery_task_id": async_result.id, "mode": mode}), 202
|
|
|
|
from ..tasks.scan import scan_directory
|
|
|
|
async_result = scan_directory.delay(triggered_by="manual", mode=mode)
|
|
return jsonify({"celery_task_id": async_result.id, "mode": mode}), 202
|
|
|
|
|
|
@import_admin_bp.route("/status", methods=["GET"])
|
|
async def status():
|
|
async with get_session() as session:
|
|
active = (
|
|
await session.execute(
|
|
select(ImportBatch)
|
|
.where(ImportBatch.status == "running")
|
|
.order_by(ImportBatch.started_at.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
payload = {"active_batch": None}
|
|
if active:
|
|
payload["active_batch"] = {
|
|
"id": active.id,
|
|
"total_files": active.total_files,
|
|
"imported": active.imported,
|
|
"skipped": active.skipped,
|
|
"failed": active.failed,
|
|
"started_at": active.started_at.isoformat(),
|
|
}
|
|
return jsonify(payload)
|
|
|
|
|
|
@import_admin_bp.route("/tasks", methods=["GET"])
|
|
async def list_tasks():
|
|
status_filter = request.args.get("status")
|
|
try:
|
|
limit = min(int(request.args.get("limit", "50")), 200)
|
|
except ValueError:
|
|
return jsonify({"error": "limit must be an integer"}), 400
|
|
cursor_raw = request.args.get("cursor")
|
|
cursor_id = int(cursor_raw) if cursor_raw else None
|
|
|
|
async with get_session() as session:
|
|
stmt = select(ImportTask).order_by(ImportTask.created_at.desc(), ImportTask.id.desc())
|
|
if status_filter:
|
|
stmt = stmt.where(ImportTask.status == status_filter)
|
|
if cursor_id is not None:
|
|
stmt = stmt.where(ImportTask.id < cursor_id)
|
|
stmt = stmt.limit(limit + 1)
|
|
rows = (await session.execute(stmt)).scalars().all()
|
|
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
return jsonify({
|
|
"tasks": [
|
|
{
|
|
"id": t.id,
|
|
"batch_id": t.batch_id,
|
|
"source_path": t.source_path,
|
|
"task_type": t.task_type,
|
|
"status": t.status,
|
|
"result_image_id": t.result_image_id,
|
|
"error": t.error,
|
|
"size_bytes": t.size_bytes,
|
|
"created_at": t.created_at.isoformat(),
|
|
"started_at": t.started_at.isoformat() if t.started_at else None,
|
|
"finished_at": t.finished_at.isoformat() if t.finished_at else None,
|
|
}
|
|
for t in rows
|
|
],
|
|
"next_cursor": rows[-1].id if has_more and rows else None,
|
|
})
|
|
|
|
|
|
@import_admin_bp.route("/retry-failed", methods=["POST"])
|
|
async def retry_failed():
|
|
async with get_session() as session:
|
|
failed_ids = (
|
|
await session.execute(select(ImportTask.id).where(ImportTask.status == "failed"))
|
|
).scalars().all()
|
|
if not failed_ids:
|
|
return jsonify({"retried": 0})
|
|
await session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.id.in_(failed_ids))
|
|
.values(status="queued", error=None, started_at=None, finished_at=None)
|
|
)
|
|
await session.commit()
|
|
|
|
from ..tasks.import_file import import_media_file
|
|
for tid in failed_ids:
|
|
import_media_file.delay(tid)
|
|
|
|
return jsonify({"retried": len(failed_ids)})
|
|
|
|
|
|
@import_admin_bp.route("/clear-stuck", methods=["POST"])
|
|
async def clear_stuck():
|
|
"""Force any non-terminal ImportTask (status in pending/queued/
|
|
processing) to 'failed' AND finalize any ImportBatch that ends up
|
|
with no active children. Escape hatch for the operator when the
|
|
automatic recover_interrupted_tasks sweep keeps re-queueing the
|
|
same stuck row forever (e.g., underlying file is genuinely broken
|
|
and the import keeps OSError-looping at PIL load).
|
|
|
|
Idempotent + non-destructive: rows survive as 'failed' so the
|
|
Retry-Failed button can re-attempt them once whatever was broken
|
|
is fixed. Banked 2026-05-25 — operator hit 3 large PNGs that
|
|
autoretry-looped for 2 days after a corrupt-data PIL OSError.
|
|
"""
|
|
async with get_session() as session:
|
|
stuck_ids = (
|
|
await session.execute(
|
|
select(ImportTask.id).where(
|
|
ImportTask.status.in_(["pending", "queued", "processing"])
|
|
)
|
|
)
|
|
).scalars().all()
|
|
if stuck_ids:
|
|
await session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.id.in_(stuck_ids))
|
|
.values(
|
|
status="failed",
|
|
finished_at=datetime.now(UTC),
|
|
error=(
|
|
"manually cleared via /api/import/clear-stuck "
|
|
"— stuck in non-terminal state; retry once "
|
|
"underlying cause (corrupt file, missing model, "
|
|
"etc.) is resolved"
|
|
),
|
|
)
|
|
)
|
|
|
|
# Finalize any 'running' ImportBatch that no longer has any
|
|
# active children. The "Scanning..." banner is driven by
|
|
# /api/import/status finding a running batch; left untouched,
|
|
# it would persist forever after the stuck-task clear.
|
|
running_batches = (
|
|
await session.execute(
|
|
select(ImportBatch.id).where(ImportBatch.status == "running")
|
|
)
|
|
).scalars().all()
|
|
finalized_batches = 0
|
|
for batch_id in running_batches:
|
|
still_active = (
|
|
await session.execute(
|
|
select(ImportTask.id)
|
|
.where(ImportTask.batch_id == batch_id)
|
|
.where(ImportTask.status.in_(
|
|
["pending", "queued", "processing"]
|
|
))
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
if still_active is None:
|
|
await session.execute(
|
|
update(ImportBatch)
|
|
.where(ImportBatch.id == batch_id)
|
|
.values(
|
|
status="complete",
|
|
finished_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
finalized_batches += 1
|
|
await session.commit()
|
|
|
|
return jsonify({
|
|
"tasks_failed": len(stuck_ids),
|
|
"batches_finalized": finalized_batches,
|
|
})
|
|
|
|
|
|
@import_admin_bp.route("/clear-completed", methods=["POST"])
|
|
async def clear_completed():
|
|
body = await request.get_json(silent=True) or {}
|
|
age_days = body.get("age_days", 0)
|
|
status_filter = body.get("status", ["complete", "skipped"])
|
|
if not isinstance(status_filter, list):
|
|
return jsonify({"error": "status must be a list"}), 400
|
|
|
|
cutoff = (
|
|
datetime.now(UTC) - timedelta(days=int(age_days)) if age_days else None
|
|
)
|
|
|
|
async with get_session() as session:
|
|
stmt = delete(ImportTask).where(ImportTask.status.in_(status_filter))
|
|
if cutoff is not None:
|
|
stmt = stmt.where(ImportTask.finished_at < cutoff)
|
|
result = await session.execute(stmt)
|
|
await session.commit()
|
|
|
|
return jsonify({"deleted": result.rowcount or 0})
|