fix(audit-g4): status-enum miss batch
Five extension-miss findings from the 2026-06-02 audit, where a status
value was added on one side but a downstream consumer didn't pick it up.
- download_service._phase3_persist: explicit branches for
ImportResult.status in ('failed','refreshed'). For 'failed' (archive
probe crash from _import_archive), unlink the source file so the
filesystem scanner doesn't re-import and re-crash on the same
archive forever. 'refreshed' is currently unreachable from the
download path (no deep=True) but matches the importer's documented
contract; treat as 'attached'.
- gallery-dl backfill auto-complete now gates on dl_result.success +
no error_type, not just return_code==0 + files_downloaded==0.
VALIDATION_FAILED exits the subprocess with returncode=0 and
files_downloaded=0 when every file was quarantined, matching the
prior predicate exactly and zeroing the operator's armed backfill
budget on the FIRST quarantine run instead of decrementing.
- attach_in_place archive dispatch now threads artist + source_row
through _import_archive (and _import_media for archive members)
and _supersede. The path-walk fallback (_resolve_artist) is still
used by filesystem-import; the download path now binds
ImageProvenance to the explicit subscription Source instead of
rediscovering by (artist_id, platform).
- Three FE handlers now recognize status:'deferred' from
/api/sources/<id>/check: SubscriptionsTab.onCheck (was toasting
"event #undefined"), SubscriptionsTab.checkAll (was counting
deferred as queued), DownloadEventRow.onRetry (was saying
"re-queued" when nothing was). Pattern matches DownloadsTab.onRetryAll
which already had it.
- celery_signals._queue_for now maps backup/admin/library_audit
prefixes to 'maintenance' (matching task_routes). TaskRun.queue
was returning 'default' for those rows, so per-queue dashboard
filters and per-queue threshold overrides (added in G3) silently
missed them.
This commit is contained in:
@@ -54,7 +54,14 @@ _INT32_MIN = -2_147_483_648
|
||||
|
||||
def _queue_for(task) -> str:
|
||||
"""Reverse the task→queue routing from celery_app.task_routes.
|
||||
Keep in sync if task_routes is reordered."""
|
||||
Keep in sync if task_routes is reordered.
|
||||
|
||||
Audit 2026-06-02: backup/admin/library_audit prefixes were
|
||||
missing here even though task_routes sent all three to
|
||||
'maintenance'. The TaskRun.queue column then lied for those
|
||||
rows (claimed 'default') so per-queue dashboard filters and
|
||||
per-queue threshold overrides silently missed them.
|
||||
"""
|
||||
name = getattr(task, "name", "") or ""
|
||||
if name.startswith("backend.app.tasks.import_file."):
|
||||
return "import"
|
||||
@@ -66,7 +73,12 @@ def _queue_for(task) -> str:
|
||||
return "download"
|
||||
if name.startswith("backend.app.tasks.scan."):
|
||||
return "scan"
|
||||
if name.startswith("backend.app.tasks.maintenance."):
|
||||
if name.startswith((
|
||||
"backend.app.tasks.maintenance.",
|
||||
"backend.app.tasks.backup.",
|
||||
"backend.app.tasks.admin.",
|
||||
"backend.app.tasks.library_audit.",
|
||||
)):
|
||||
return "maintenance"
|
||||
return "default"
|
||||
|
||||
|
||||
@@ -315,6 +315,23 @@ class DownloadService:
|
||||
# failure. Don't flag the run as error; the file stays
|
||||
# on disk for operator inspection.
|
||||
import_summary["skipped"] += 1
|
||||
elif result.status == "failed":
|
||||
# Hard failure (today only: archive probe crash/timeout).
|
||||
# The original archive sits in /images/ as an orphan; the
|
||||
# filesystem scanner would re-import and re-crash on the
|
||||
# same file, so delete the source file and surface the
|
||||
# error in import_summary. Audit 2026-06-02.
|
||||
import_summary["errors"] += 1
|
||||
try:
|
||||
path.unlink(missing_ok=True) # noqa: ASYNC240
|
||||
except OSError:
|
||||
pass
|
||||
elif result.status == "refreshed":
|
||||
# Currently unreachable from attach_in_place (the download
|
||||
# path never runs in deep=True mode), but the importer's
|
||||
# ImportResult contract enumerates it. Treat the same as
|
||||
# 'attached' — work happened, no error. Audit 2026-06-02.
|
||||
import_summary["attached"] += 1
|
||||
else:
|
||||
import_summary["errors"] += 1
|
||||
|
||||
@@ -361,12 +378,26 @@ class DownloadService:
|
||||
# backfill run drained the queue (gallery-dl exited 0 + zero files
|
||||
# downloaded means there was nothing to fetch); otherwise decrement
|
||||
# the counter. Next tick falls back to tick mode once it hits 0.
|
||||
#
|
||||
# Audit 2026-06-02 gating: VALIDATION_FAILED also exits the
|
||||
# subprocess with return_code=0 and files_downloaded=0 (every
|
||||
# file was quarantined), which used to match the auto-complete
|
||||
# predicate exactly — zeroing the operator's armed budget on
|
||||
# the FIRST quarantine run instead of decrementing. Require
|
||||
# dl_result.success + no error_type so only genuinely-empty
|
||||
# successful runs drain the counter.
|
||||
backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
|
||||
if backfill_remaining > 0:
|
||||
src = (await self.async_session.execute(
|
||||
select(Source).where(Source.id == ctx["source_id"])
|
||||
)).scalar_one()
|
||||
if dl_result.return_code == 0 and dl_result.files_downloaded == 0:
|
||||
queue_drained = (
|
||||
dl_result.success
|
||||
and dl_result.error_type is None
|
||||
and dl_result.return_code == 0
|
||||
and dl_result.files_downloaded == 0
|
||||
)
|
||||
if queue_drained:
|
||||
src.backfill_runs_remaining = 0
|
||||
else:
|
||||
src.backfill_runs_remaining = max(0, backfill_remaining - 1)
|
||||
|
||||
@@ -393,7 +393,11 @@ class Importer:
|
||||
self.session.commit()
|
||||
return ImportResult(status="attached")
|
||||
|
||||
def _import_archive(self, source: Path) -> ImportResult:
|
||||
def _import_archive(
|
||||
self, source: Path, *,
|
||||
artist: Artist | None = None,
|
||||
source_row: Source | None = None,
|
||||
) -> ImportResult:
|
||||
# Layer-3 isolation: bomb-size guard + integrity test in a
|
||||
# spawned child BEFORE extracting in this process. A
|
||||
# decompression bomb or a native-lib crash on a malformed
|
||||
@@ -401,6 +405,14 @@ class Importer:
|
||||
# instead of OOMing/segfaulting the import worker. extract_archive
|
||||
# is already fail-soft for plain exceptions, so this only adds
|
||||
# the hard-crash protection.
|
||||
#
|
||||
# Audit 2026-06-02: optional artist/source_row kwargs let the
|
||||
# download path thread its explicit subscription context
|
||||
# through instead of having _resolve_artist re-derive from
|
||||
# path-walk (which works by coincidence today because gallery-dl
|
||||
# lays files out under /images/<artist_slug>/...). Filesystem
|
||||
# import still calls bare _import_archive(source) and falls
|
||||
# back to the path-walk derivation as before.
|
||||
probe = safe_probe.probe_archive(source)
|
||||
if not probe.ok:
|
||||
if probe.crashed:
|
||||
@@ -412,24 +424,28 @@ class Importer:
|
||||
# still preserve the archive file itself as an attachment so
|
||||
# nothing silently vanishes, matching extract_archive's
|
||||
# fail-soft contract.
|
||||
artist = self._resolve_artist(source)
|
||||
post = self._post_for_sidecar(source, artist)
|
||||
self._capture_attachment(source, post=post, artist=artist, resolved=True)
|
||||
artist_use = artist if artist is not None else self._resolve_artist(source)
|
||||
post = self._post_for_sidecar(source, artist_use)
|
||||
self._capture_attachment(
|
||||
source, post=post, artist=artist_use, resolved=True,
|
||||
)
|
||||
return ImportResult(status="attached")
|
||||
|
||||
artist = self._resolve_artist(source)
|
||||
post = self._post_for_sidecar(source, artist)
|
||||
artist_use = artist if artist is not None else self._resolve_artist(source)
|
||||
post = self._post_for_sidecar(source, artist_use)
|
||||
member_ids: list[int] = []
|
||||
with extract_archive(source) as members:
|
||||
for _name, member_path in members:
|
||||
if not is_supported(member_path):
|
||||
continue # non-media preserved via the stored archive
|
||||
res = self._import_media(member_path, source)
|
||||
res = self._import_media(
|
||||
member_path, source, explicit_source=source_row,
|
||||
)
|
||||
if res.status in ("imported", "superseded") and res.image_id:
|
||||
member_ids.append(res.image_id)
|
||||
# Preserve the archive itself (links to the same Post/Artist).
|
||||
self._capture_attachment(
|
||||
source, post=post, artist=artist, resolved=True
|
||||
source, post=post, artist=artist_use, resolved=True
|
||||
)
|
||||
if member_ids:
|
||||
return ImportResult(
|
||||
@@ -439,7 +455,8 @@ class Importer:
|
||||
return ImportResult(status="attached")
|
||||
|
||||
def _import_media(
|
||||
self, source: Path, attribution_path: Path
|
||||
self, source: Path, attribution_path: Path,
|
||||
*, explicit_source: Source | None = None,
|
||||
) -> ImportResult:
|
||||
"""The media import pipeline (filters, dedup, copy, provenance).
|
||||
|
||||
@@ -586,7 +603,15 @@ class Importer:
|
||||
artist = self._attach_artist(record, artist_name)
|
||||
|
||||
# Sidecar provenance (best-effort; never fails the import).
|
||||
self._apply_sidecar(record, attribution_path, artist)
|
||||
# explicit_source lets the FC-3c download path bind the new
|
||||
# ImageProvenance row to its subscription Source instead of
|
||||
# having _apply_sidecar re-derive via _lookup_source_for_sidecar.
|
||||
# Audit 2026-06-02 — archive members extracted from a
|
||||
# subscription-downloaded zip previously lost subscription
|
||||
# linkage if the on-disk layout didn't match assumptions.
|
||||
self._apply_sidecar(
|
||||
record, attribution_path, artist, explicit_source=explicit_source,
|
||||
)
|
||||
|
||||
# Thumbnail is queued separately by the calling task; the importer
|
||||
# does not generate thumbnails inline so the import queue stays moving.
|
||||
@@ -673,7 +698,9 @@ class Importer:
|
||||
error="sidecar json is metadata, not content",
|
||||
)
|
||||
if is_archive(path):
|
||||
return self._import_archive(path)
|
||||
return self._import_archive(
|
||||
path, artist=artist, source_row=source,
|
||||
)
|
||||
if not is_supported(path):
|
||||
post = self._post_for_sidecar(path, artist) if artist else None
|
||||
return self._capture_attachment(
|
||||
@@ -749,7 +776,8 @@ class Importer:
|
||||
if rel == "smaller_exists":
|
||||
target = self.session.get(ImageRecord, match_id)
|
||||
self._supersede(
|
||||
target, path, sha, phash, width, height, new_path=path
|
||||
target, path, sha, phash, width, height,
|
||||
new_path=path, artist=artist, source_row=source,
|
||||
)
|
||||
return ImportResult(status="superseded", image_id=match_id)
|
||||
|
||||
@@ -944,6 +972,8 @@ class Importer:
|
||||
self, existing: ImageRecord, source: Path, sha: str,
|
||||
phash: str, width: int | None, height: int | None,
|
||||
*, new_path: Path | None = None,
|
||||
artist: Artist | None = None,
|
||||
source_row: Source | None = None,
|
||||
) -> None:
|
||||
"""Replace `existing`'s file with the larger `source`, keeping the
|
||||
row id (so tags/series/curation stay attached). ML is cleared so
|
||||
@@ -995,8 +1025,14 @@ class Importer:
|
||||
# _apply_sidecar resolves artist from the sidecar itself if the
|
||||
# existing row has none, and is internally guarded against
|
||||
# missing-or-malformed sidecars (silent return).
|
||||
# Audit 2026-06-02: thread artist/source_row from the
|
||||
# download-path caller (attach_in_place smaller_exists branch)
|
||||
# so the supersede preserves explicit subscription linkage
|
||||
# instead of re-deriving via path-walk.
|
||||
try:
|
||||
self._apply_sidecar(existing, source, None)
|
||||
self._apply_sidecar(
|
||||
existing, source, artist, explicit_source=source_row,
|
||||
)
|
||||
except Exception as exc:
|
||||
# Don't unwind the supersede DB swap if sidecar parsing
|
||||
# blows up unexpectedly — the file replacement is the
|
||||
|
||||
@@ -124,10 +124,16 @@ async function onRetry() {
|
||||
if (!props.event.source_id) return
|
||||
retrying.value = true
|
||||
try {
|
||||
await sourcesStore.checkNow(props.event.source_id)
|
||||
toast({
|
||||
text: `Source check re-queued`, type: 'success',
|
||||
})
|
||||
const body = await sourcesStore.checkNow(props.event.source_id)
|
||||
// Audit 2026-06-02: the previous handler unconditionally toasted
|
||||
// "re-queued" even when the platform was in cooldown (202 +
|
||||
// status='deferred'). Operator thought work was in flight when
|
||||
// nothing was actually enqueued.
|
||||
if (body?.status === 'deferred') {
|
||||
toast({ text: 'Retry deferred — platform in cooldown', type: 'info' })
|
||||
} else {
|
||||
toast({ text: 'Source check re-queued', type: 'success' })
|
||||
}
|
||||
} catch (e) {
|
||||
const isInFlight = !!e?.body?.download_event_id
|
||||
toast({
|
||||
|
||||
@@ -412,6 +412,17 @@ function onArtistCreated(artist) {
|
||||
async function onCheck(source) {
|
||||
try {
|
||||
const body = await store.checkNow(source.id)
|
||||
// Audit 2026-06-02: /api/sources/<id>/check returns 202 with
|
||||
// `{status:'deferred', cooldown_until}` when the platform is in
|
||||
// cooldown — the previous handler treated this as success and
|
||||
// toasted "event #undefined", masking that nothing was enqueued.
|
||||
if (body?.status === 'deferred') {
|
||||
toast({
|
||||
text: 'Check deferred — platform in cooldown',
|
||||
type: 'info',
|
||||
})
|
||||
return
|
||||
}
|
||||
toast({
|
||||
text: `Check enqueued (event #${body.download_event_id})`,
|
||||
type: 'success',
|
||||
@@ -465,17 +476,23 @@ async function onBackfill(source) {
|
||||
async function checkAll(group) {
|
||||
let ok = 0
|
||||
let conflict = 0
|
||||
let deferred = 0
|
||||
for (const s of group.sources) {
|
||||
if (!s.enabled) continue
|
||||
try {
|
||||
await store.checkNow(s.id)
|
||||
ok += 1
|
||||
const body = await store.checkNow(s.id)
|
||||
// Audit 2026-06-02: deferred (202 + cooldown_until) used to be
|
||||
// counted as queued, inflating the success tally and hiding
|
||||
// that the cooldown actually held the work back.
|
||||
if (body?.status === 'deferred') deferred += 1
|
||||
else ok += 1
|
||||
} catch (e) {
|
||||
if (e?.body?.download_event_id) conflict += 1
|
||||
}
|
||||
}
|
||||
const parts = []
|
||||
if (ok) parts.push(`${ok} queued`)
|
||||
if (deferred) parts.push(`${deferred} deferred (cooldown)`)
|
||||
if (conflict) parts.push(`${conflict} already running`)
|
||||
toast({
|
||||
text: parts.join(', ') || 'Nothing to check (no enabled sources)',
|
||||
|
||||
Reference in New Issue
Block a user