feat(import-resilience L2): one-shot re-download for corrupt downloaded files

Layer 2 — remediate a corrupt file by re-fetching a fresh copy from its
source, bounded to a single attempt. Operator-requested 2026-05-28.

New backend/app/services/refetch_service.py:
- resolve_refetch_source: parse the failed file's sidecar → platform,
  derive the artist from the import path, find an ENABLED Source with a
  real feed URL for (artist, platform). Returns None for filesystem-only
  imports, missing sidecars, or `sidecar:<platform>:<slug>` synthetic
  anchors (not pollable).
- attempt_refetch: if not already refetched AND a Source resolves,
  delete the corrupt file (so gallery-dl's skip_existing re-fetches it),
  set ImportTask.refetched=True, and trigger ONE download_source
  re-check. Bounded by `refetched` so source-side corruption can't loop.

Wiring:
- Manual endpoint POST /api/import/tasks/<id>/refetch (only on 'failed'
  tasks). Returns refetch_queued / no_source / already_refetched /
  not_found / not_failed.
- Auto path in recover_interrupted_tasks: for each poison-pill row, if
  env FC_AUTO_REFETCH_CORRUPT=1, attempt_refetch (default OFF — the
  manual button is the primary path; auto is opt-in since re-fetch
  deletes a file + re-runs the downloader).
- Frontend: a cloud-refresh icon button on failed rows in ImportTaskList
  → stores.import.refetchTask → toast keyed on the result status.

Filesystem imports with no upstream return no_source — the operator's
only remediation there is replacing the file on disk, surfaced clearly
in the toast.

Tests: 404 unknown task, 400 non-failed task, no_source when
unresolvable, and the full resolvable-source path (file deleted,
refetched flag set, one download_source dispatched, second call is a
no-op). The resolvable test repoints the migration-seeded
import_settings(id=1) scan path rather than inserting a conflicting row.
This commit is contained in:
2026-05-28 00:08:03 -04:00
parent e3cdd0f92b
commit dcfe55d731
6 changed files with 320 additions and 2 deletions
+37
View File
@@ -128,6 +128,43 @@ async def retry_failed():
return jsonify({"retried": len(failed)}) return jsonify({"retried": len(failed)})
@import_admin_bp.route("/tasks/<int:task_id>/refetch", methods=["POST"])
async def refetch_task(task_id: int):
"""Layer-2 one-shot re-download: delete the (corrupt) file behind a
failed import task and re-run its source's downloader to fetch a
fresh copy. Only works for files that resolve to an enabled,
real-URL subscription Source; filesystem-only imports return
no_source.
Returns one of: refetch_queued (+source_id) / no_source /
already_refetched / not_found / not_failed.
"""
async with get_session() as session:
result = await session.run_sync(_refetch_task_sync, task_id)
if result["status"] == "not_found":
return jsonify(result), 404
if result["status"] == "not_failed":
return jsonify(result), 400
return jsonify(result)
def _refetch_task_sync(session, task_id: int) -> dict:
from pathlib import Path
from ..models import ImportSettings
from ..services.refetch_service import attempt_refetch
task = session.get(ImportTask, task_id)
if task is None:
return {"status": "not_found"}
if task.status != "failed":
return {"status": "not_failed"}
settings = session.execute(
select(ImportSettings).where(ImportSettings.id == 1)
).scalar_one()
return attempt_refetch(session, task, Path(settings.import_scan_path))
@import_admin_bp.route("/clear-stuck", methods=["POST"]) @import_admin_bp.route("/clear-stuck", methods=["POST"])
async def clear_stuck(): async def clear_stuck():
"""Force any non-terminal ImportTask (status in pending/queued/ """Force any non-terminal ImportTask (status in pending/queued/
+106
View File
@@ -0,0 +1,106 @@
"""Layer-2 one-shot re-download remediation for corrupt imported files.
When an import fails on a file that came from a known, pollable
subscription Source, deleting the bad copy and re-running the source's
downloader can fetch a fresh, unblemished copy. This only helps when:
- the corruption is in transit / on disk (not at the source), AND
- the file resolves to an ENABLED Source with a real feed URL
(a `sidecar:<platform>:<slug>` synthetic anchor is not pollable),
AND
- we haven't already re-fetched this task once (bounded by
ImportTask.refetched so source-side corruption can't loop).
Filesystem-only imports with no resolvable Source return 'no_source'
the operator's only remediation there is to replace the file on disk.
Operator-requested 2026-05-28 (Layer 2).
"""
import json
import logging
from pathlib import Path
from sqlalchemy import select
from sqlalchemy.orm import Session
from ..models import Artist, ImportTask, Source
from ..utils.paths import derive_top_level_artist
from ..utils.sidecar import find_sidecar, parse_sidecar
from ..utils.slug import slugify
log = logging.getLogger(__name__)
def resolve_refetch_source(
session: Session, source_path: str, import_root: Path,
) -> Source | None:
"""Find an enabled, real-URL Source for the file's (artist, platform),
or None when nothing re-pollable resolves."""
path = Path(source_path)
sc = find_sidecar(path)
if sc is None:
return None
try:
data = json.loads(sc.read_text("utf-8"))
except (OSError, json.JSONDecodeError):
return None
if not isinstance(data, dict):
return None
sd = parse_sidecar(data)
if not sd.platform:
return None
artist_name = derive_top_level_artist(path, import_root)
if not artist_name:
return None
artist = session.execute(
select(Artist).where(Artist.slug == slugify(artist_name))
).scalar_one_or_none()
if artist is None:
return None
src = session.execute(
select(Source)
.where(
Source.artist_id == artist.id,
Source.platform == sd.platform,
Source.enabled.is_(True),
)
.order_by(Source.id.asc())
).scalars().first()
if src is None:
return None
if (src.url or "").startswith("sidecar:"):
return None # synthetic anchor — not a pollable feed
return src
def attempt_refetch(
session: Session, task: ImportTask, import_root: Path,
) -> dict:
"""Delete the corrupt file, mark the task refetched, and trigger ONE
source re-check. Idempotent/bounded: a task already refetched (or
with no resolvable Source) is a no-op. Commits."""
if task.refetched:
return {"status": "already_refetched"}
src = resolve_refetch_source(session, task.source_path, import_root)
if src is None:
return {"status": "no_source"}
# Remove the bad copy so gallery-dl (skip_existing) re-fetches it on
# the source re-check instead of skipping the still-present corrupt
# file.
try:
Path(task.source_path).unlink(missing_ok=True)
except OSError as exc:
log.warning("refetch unlink failed for %s: %s", task.source_path, exc)
task.refetched = True
session.add(task)
session.commit()
# Lazy import to avoid a tasks→services→tasks import cycle at module
# load. download_source.delay() is sync-safe in any context.
from ..tasks.download import download_source
download_source.delay(src.id)
return {"status": "refetch_queued", "source_id": src.id}
+23
View File
@@ -1,6 +1,7 @@
"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks.""" """Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
import logging import logging
import os
import subprocess import subprocess
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from pathlib import Path from pathlib import Path
@@ -183,6 +184,28 @@ def recover_interrupted_tasks() -> int:
for tid, task_type in stuck: for tid, task_type in stuck:
enqueue_import(tid, task_type) enqueue_import(tid, task_type)
# Layer-2 auto re-download (env-gated, default OFF). For each
# poison-pill row that resolves to a pollable Source, delete the
# bad file and trigger ONE source re-check to fetch a fresh
# copy. Bounded by ImportTask.refetched so source-side
# corruption can't loop. The 'failed' row stays as history; the
# re-downloaded file re-imports as a fresh task on the next scan.
if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1":
from ..models import ImportSettings
from ..services.refetch_service import attempt_refetch
import_root = Path(session.execute(
select(ImportSettings.import_scan_path)
.where(ImportSettings.id == 1)
).scalar_one())
for pid in poison_ids:
ptask = session.get(ImportTask, pid)
if ptask is None:
continue
try:
attempt_refetch(session, ptask, import_root)
except Exception as exc: # noqa: BLE001 — best-effort
log.warning("auto-refetch failed for task %s: %s", pid, exc)
return len(stuck) + len(poison_ids) + orphan_count return len(stuck) + len(poison_ids) + orphan_count
@@ -51,6 +51,19 @@
title="Click for full error" title="Click for full error"
>{{ shorten(item.error, 60) }}</button> >{{ shorten(item.error, 60) }}</button>
</template> </template>
<template #item.actions="{ item }">
<v-btn
v-if="item.status === 'failed'"
icon size="x-small" variant="text"
:loading="refetching === item.id"
@click="onRefetch(item)"
>
<v-icon size="small">mdi-cloud-refresh</v-icon>
<v-tooltip activator="parent" location="top">
Re-fetch original (re-download from source)
</v-tooltip>
</v-btn>
</template>
</v-data-table-virtual> </v-data-table-virtual>
<div v-if="store.hasMore" class="d-flex justify-center py-3"> <div v-if="store.hasMore" class="d-flex justify-center py-3">
<v-btn variant="text" size="small" @click="onLoadMore">Load more</v-btn> <v-btn variant="text" size="small" @click="onLoadMore">Load more</v-btn>
@@ -149,9 +162,29 @@ const headers = [
{ title: 'Source', key: 'source_path', sortable: false }, { title: 'Source', key: 'source_path', sortable: false },
{ title: 'Size', key: 'size_bytes', sortable: false, width: 90 }, { title: 'Size', key: 'size_bytes', sortable: false, width: 90 },
{ title: 'Created', key: 'created_at', sortable: false, width: 150 }, { title: 'Created', key: 'created_at', sortable: false, width: 150 },
{ title: 'Note', key: 'error', sortable: false } { title: 'Note', key: 'error', sortable: false },
{ title: '', key: 'actions', sortable: false, width: 56 }
] ]
const refetching = ref(null)
const _REFETCH_MSG = {
refetch_queued: { text: 'Re-fetch queued — re-downloading from source', type: 'success' },
no_source: { text: 'No re-fetchable source (filesystem import — replace the file manually)', type: 'info' },
already_refetched: { text: 'Already re-fetched once', type: 'info' },
}
async function onRefetch(item) {
refetching.value = item.id
try {
const res = await store.refetchTask(item.id)
const msg = _REFETCH_MSG[res.status] || { text: `Re-fetch: ${res.status}`, type: 'info' }
window.__fcToast?.(msg)
} catch (e) {
window.__fcToast?.({ text: `Re-fetch failed: ${e.message}`, type: 'error' })
} finally {
refetching.value = null
}
}
const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed')) const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed'))
const hasStuck = computed(() => store.tasks.some( const hasStuck = computed(() => store.tasks.some(
t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing' t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing'
+11 -1
View File
@@ -146,6 +146,15 @@ export const useImportStore = defineStore('import', () => {
return body return body
} }
// Layer-2 one-shot re-download for a failed task's (corrupt) file.
// Returns the endpoint's status dict (refetch_queued / no_source /
// already_refetched). Caller surfaces it as a toast.
async function refetchTask(taskId) {
const body = await api.post(`/api/import/tasks/${taskId}/refetch`)
await loadTasks(true)
return body
}
const hasMore = computed(() => tasksNextCursor.value !== null) const hasMore = computed(() => tasksNextCursor.value !== null)
return { return {
@@ -155,6 +164,7 @@ export const useImportStore = defineStore('import', () => {
triggerError, triggerError,
loadSettings, patchSettings, loadSettings, patchSettings,
refreshStatus, triggerScan, refreshStatus, triggerScan,
loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck,
refetchTask,
} }
}) })
+109
View File
@@ -171,6 +171,115 @@ async def test_trigger_still_rejects_unknown_mode(client):
assert resp.status_code == 400 assert resp.status_code == 400
@pytest.mark.asyncio
async def test_refetch_404_for_unknown_task(client):
resp = await client.post("/api/import/tasks/999999/refetch")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_refetch_400_for_non_failed_task(client, db):
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
db.add(batch)
await db.flush()
task = ImportTask(
batch_id=batch.id, source_path="/x.jpg", task_type="media",
status="complete", finished_at=datetime.now(UTC),
)
db.add(task)
await db.commit()
resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
assert resp.status_code == 400
assert (await resp.get_json())["status"] == "not_failed"
@pytest.mark.asyncio
async def test_refetch_no_source_when_unresolvable(client, db):
"""A failed task whose file has no sidecar / no resolvable Source
returns no_source (filesystem-only import — nothing to re-poll)."""
batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
db.add(batch)
await db.flush()
task = ImportTask(
batch_id=batch.id, source_path="/import/nowhere/x.jpg",
task_type="media", status="failed", finished_at=datetime.now(UTC),
)
db.add(task)
await db.commit()
resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
assert resp.status_code == 200
assert (await resp.get_json())["status"] == "no_source"
@pytest.mark.asyncio
async def test_refetch_queued_with_resolvable_source(client, db, tmp_path, monkeypatch):
"""A failed task whose file resolves (via sidecar → artist+platform)
to an enabled, real-URL Source: the file is deleted, the task is
marked refetched, and ONE source re-check is queued."""
import json as _json
from sqlalchemy import update as _update
from backend.app.models import Artist, ImportSettings, Source
from backend.app.tasks import download as download_mod
# Stub the downloader so the eager test doesn't run a real fetch.
dispatched = []
monkeypatch.setattr(download_mod.download_source, "delay", dispatched.append)
# import_root/<ArtistName>/post.jpg + sidecar identifying the platform.
import_root = tmp_path / "import"
artist_dir = import_root / "Maewix"
artist_dir.mkdir(parents=True)
media = artist_dir / "post.jpg"
media.write_bytes(b"corrupt-bytes")
(artist_dir / "post.jpg.json").write_text(
_json.dumps({"category": "patreon", "post_id": 123})
)
# import_settings(id=1) is migration-seeded; point its scan path at
# our tmp import root rather than inserting a conflicting row.
await db.execute(
_update(ImportSettings).where(ImportSettings.id == 1)
.values(import_scan_path=str(import_root))
)
artist = Artist(name="Maewix", slug="maewix")
db.add(artist)
await db.flush()
db.add(Source(
artist_id=artist.id, platform="patreon",
url="https://www.patreon.com/maewix", enabled=True,
config_overrides={},
))
batch = ImportBatch(triggered_by="manual", source_path=str(import_root), scan_mode="quick")
db.add(batch)
await db.flush()
task = ImportTask(
batch_id=batch.id, source_path=str(media), task_type="media",
status="failed", finished_at=datetime.now(UTC),
)
db.add(task)
await db.commit()
resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
assert resp.status_code == 200
body = await resp.get_json()
assert body["status"] == "refetch_queued"
assert len(dispatched) == 1
assert not media.exists() # corrupt copy removed for re-fetch
from sqlalchemy import select as _select
refetched = (await db.execute(
_select(ImportTask.refetched).where(ImportTask.id == task.id)
)).scalar_one()
assert refetched is True
# Second attempt is a no-op (bounded to one).
resp2 = await client.post(f"/api/import/tasks/{task.id}/refetch")
assert (await resp2.get_json())["status"] == "already_refetched"
assert len(dispatched) == 1
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_trigger_accepts_verify(client, monkeypatch): async def test_trigger_accepts_verify(client, monkeypatch):
# Stub the verify task's dispatch so the API contract is asserted # Stub the verify task's dispatch so the API contract is asserted