diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index 37bd482..222a565 100644 --- a/backend/app/celery_signals.py +++ b/backend/app/celery_signals.py @@ -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" diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index f58bb3d..d8c15e9 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -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) diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index fb39052..13614c3 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -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//...). 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 diff --git a/frontend/src/components/downloads/DownloadEventRow.vue b/frontend/src/components/downloads/DownloadEventRow.vue index 8a5a6fb..3c96702 100644 --- a/frontend/src/components/downloads/DownloadEventRow.vue +++ b/frontend/src/components/downloads/DownloadEventRow.vue @@ -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({ diff --git a/frontend/src/components/subscriptions/SubscriptionsTab.vue b/frontend/src/components/subscriptions/SubscriptionsTab.vue index eb80cc1..73cbb0b 100644 --- a/frontend/src/components/subscriptions/SubscriptionsTab.vue +++ b/frontend/src/components/subscriptions/SubscriptionsTab.vue @@ -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//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)',