bvandeusen
593f65c9cc
Large Patreon creators (Anduo: weekly 50-120-image Reports back months = thousands of files) couldn't backfill: each run re-walked newest→oldest from the top, and gallery-dl's polite ~0.75s/request HEAD walk alone exceeded the 1170s subprocess budget, so the run died during enumeration with 0 files written and NO forward progress — re-stranding every time (event #40411). Checkpoint gallery-dl's pagination cursor so each backfill window advances the frontier: - gallery_dl.py: SourceConfig.resume_cursor; _build_config_for_source sets extractor.patreon.cursor=<resume> (PLATFORM_DEFAULTS leave log-only True for a fresh run); parse_last_cursor() pulls the last emitted 'Cursor: <token>' from stdout+stderr — survives a timed-out run since the TimeoutExpired path returns partial output. - download_service.py: phase2 stays in BACKFILL mode while a cursor is pending (even after the run budget drains) and threads resume_cursor; _apply_backfill_lifecycle() checkpoints the advancing cursor each non-completing run, completes on a clean rc=0 finish (walk reached bottom), and a stuck-guard clears the cursor after 2 non-advancing runs so a wedged walk can't re-strand forever. patreon-only (sole platform with a resumable cursor); other platforms keep the simple counter semantics. Cursor state lives in config_overrides JSON (patreon_campaign_id precedent) — no migration. Time-budget ladder (1170/1350/1500) unchanged. Plan #689. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
535 lines
24 KiB
Python
535 lines
24 KiB
Python
"""FC-3c download orchestrator.
|
|
|
|
Three-phase pipeline:
|
|
Phase 1 — DB setup (brief session): load source, mark event running,
|
|
fetch credentials + settings.
|
|
Phase 2 — Execute (no DB connection): call GalleryDLService; on
|
|
Patreon campaign-ID failure, resolve vanity and retry once.
|
|
Phase 3 — Import + persist (fresh session): attach_in_place each
|
|
written file; clean up duplicates; persist rich metadata.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Session as SyncSession
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from ..models import Artist, DownloadEvent, Source
|
|
from .credential_service import CredentialService
|
|
from .gallery_dl import (
|
|
BACKFILL_SKIP_VALUE,
|
|
BACKFILL_TIMEOUT_SECONDS,
|
|
TICK_SKIP_VALUE,
|
|
ErrorType,
|
|
GalleryDLService,
|
|
SourceConfig,
|
|
parse_last_cursor,
|
|
)
|
|
from .importer import Importer
|
|
from .patreon_resolver import resolve_campaign_id
|
|
from .platforms import auth_type_for
|
|
from .scheduler_service import set_platform_cooldown
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
_PATREON_VANITY_RE = re.compile(
|
|
r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
|
|
re.IGNORECASE,
|
|
)
|
|
_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"
|
|
|
|
|
|
def _extract_patreon_vanity(url: str) -> str | None:
|
|
m = _PATREON_VANITY_RE.match(url)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def _looks_like_campaign_id_failure(stdout: str, stderr: str) -> bool:
|
|
return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()
|
|
|
|
|
|
def _effective_url(platform: str, source_url: str, overrides: dict) -> str:
|
|
if platform == "patreon" and overrides.get("patreon_campaign_id"):
|
|
return f"https://www.patreon.com/id:{overrides['patreon_campaign_id']}"
|
|
return source_url
|
|
|
|
|
|
class DownloadService:
|
|
"""Async orchestrator. The Celery task runs `asyncio.run(svc.download_source(N))`.
|
|
|
|
The Importer is sync (existing FC convention); calls to attach_in_place
|
|
happen via run_in_executor.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
async_session: AsyncSession,
|
|
sync_session: SyncSession,
|
|
gdl: GalleryDLService,
|
|
importer: Importer,
|
|
cred_service: CredentialService,
|
|
):
|
|
self.async_session = async_session
|
|
self.sync_session = sync_session
|
|
self.gdl = gdl
|
|
self.importer = importer
|
|
self.cred_service = cred_service
|
|
|
|
async def download_source(self, source_id: int) -> int:
|
|
"""Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is."""
|
|
setup = await self._phase1_setup(source_id)
|
|
if setup["status"] in ("skipped", "in_flight"):
|
|
return setup["event_id"]
|
|
ctx = setup
|
|
|
|
# Release the phase-1 DB connections before the (up to ~19.5-min in
|
|
# backfill) gallery-dl subprocess. Held checked-out across that idle
|
|
# window, the asyncpg/psycopg connections get reaped by the server,
|
|
# and phase 3's first query then hits a dead socket
|
|
# (asyncpg ConnectionDoesNotExistError) → download_source autoretry →
|
|
# _phase1_setup's in-flight guard no-ops the retry → the event
|
|
# strands empty for the recovery sweep (Anduo #40014, 2026-06-04).
|
|
# pool_pre_ping can't help a *held* connection — it only validates on
|
|
# pool checkout. Closing returns them to the pool so phase 3 re-
|
|
# acquires a live one (the async task engine uses NullPool, the sync
|
|
# engine pre_ping + pool_recycle=300). This is what makes the
|
|
# "Phase 2 — no DB connection" contract in the class docstring true.
|
|
await self.async_session.close()
|
|
self.sync_session.close()
|
|
|
|
source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
|
|
# alembic 0031 / plan #544: derive skip_value + timeout from the
|
|
# source's backfill_runs_remaining counter. When > 0, walk the full
|
|
# post history (skip: True + 1170s); when 0, exit gallery-dl after
|
|
# 20 contiguous archived items (skip: "exit:20" + the default
|
|
# 870s). Operator sets backfill via POST /api/sources/{id}/backfill.
|
|
overrides = ctx["config_overrides"] or {}
|
|
backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
|
|
# Cursor-paged backfill (plan #689): a pending checkpoint keeps the
|
|
# deep-walk in backfill mode across ticks until it reaches the
|
|
# bottom, even after the operator's run budget hits 0 — so one large
|
|
# creator (Anduo) finishes without re-triggering. The phase-3
|
|
# stuck-guard bounds it. patreon-only: it's the sole platform with a
|
|
# resumable gallery-dl cursor.
|
|
pending_cursor = overrides.get("_backfill_cursor")
|
|
if backfill_remaining > 0 or pending_cursor:
|
|
skip_value: bool | str = BACKFILL_SKIP_VALUE
|
|
source_config.timeout = BACKFILL_TIMEOUT_SECONDS
|
|
if ctx["platform"] == "patreon" and pending_cursor:
|
|
source_config.resume_cursor = pending_cursor
|
|
else:
|
|
skip_value = TICK_SKIP_VALUE
|
|
|
|
effective_url = _effective_url(
|
|
ctx["platform"], ctx["url"], ctx["config_overrides"] or {}
|
|
)
|
|
|
|
dl_result = await self.gdl.download(
|
|
url=effective_url,
|
|
artist_slug=ctx["artist_slug"],
|
|
platform=ctx["platform"],
|
|
source_config=source_config,
|
|
cookies_path=ctx["cookies_path"],
|
|
auth_token=ctx["auth_token"],
|
|
skip_value=skip_value,
|
|
)
|
|
|
|
resolved_campaign_id: str | None = None
|
|
if (
|
|
ctx["platform"] == "patreon"
|
|
and not dl_result.success
|
|
and "patreon_campaign_id" not in (ctx["config_overrides"] or {})
|
|
and _looks_like_campaign_id_failure(dl_result.stdout, dl_result.stderr)
|
|
):
|
|
vanity = _extract_patreon_vanity(ctx["url"])
|
|
if vanity:
|
|
log.info(
|
|
"Attempting campaign-ID resolution for %s (%s)",
|
|
ctx["artist_slug"], vanity,
|
|
)
|
|
resolved_campaign_id = await resolve_campaign_id(
|
|
vanity, ctx["cookies_path"]
|
|
)
|
|
if resolved_campaign_id:
|
|
dl_result = await self.gdl.download(
|
|
url=f"https://www.patreon.com/id:{resolved_campaign_id}",
|
|
artist_slug=ctx["artist_slug"],
|
|
platform=ctx["platform"],
|
|
source_config=source_config,
|
|
cookies_path=ctx["cookies_path"],
|
|
auth_token=ctx["auth_token"],
|
|
skip_value=skip_value,
|
|
)
|
|
|
|
return await self._phase3_persist(
|
|
ctx["event_id"], ctx, dl_result, resolved_campaign_id,
|
|
)
|
|
|
|
async def _phase1_setup(self, source_id: int) -> dict[str, Any]:
|
|
source = (await self.async_session.execute(
|
|
select(Source).options(joinedload(Source.artist)).where(Source.id == source_id)
|
|
)).scalar_one_or_none()
|
|
if source is None:
|
|
raise LookupError(f"source {source_id} not found")
|
|
|
|
if not source.enabled:
|
|
ev = DownloadEvent(source_id=source_id, status="pending")
|
|
self.async_session.add(ev)
|
|
await self.async_session.commit()
|
|
await self.async_session.refresh(ev)
|
|
await self._finalize_event_and_source(
|
|
event_id=ev.id, source_id=source_id, status="skipped",
|
|
error_message="source disabled",
|
|
)
|
|
return {"status": "skipped", "event_id": ev.id}
|
|
|
|
existing = (await self.async_session.execute(
|
|
select(DownloadEvent).where(
|
|
DownloadEvent.source_id == source_id,
|
|
DownloadEvent.status.in_(["pending", "running"]),
|
|
).order_by(DownloadEvent.id.desc()).limit(1)
|
|
)).scalar_one_or_none()
|
|
|
|
if existing and existing.status == "running":
|
|
return {"status": "in_flight", "event_id": existing.id}
|
|
if existing and existing.status == "pending":
|
|
existing.status = "running"
|
|
# Reset started_at on the pending→running transition so the
|
|
# recovery sweep (DOWNLOAD_STALL_THRESHOLD_MINUTES, 30 min)
|
|
# measures from real start, not from enqueue. On heavy-queue
|
|
# days a freshly-promoted event whose original started_at
|
|
# predated the cutoff would otherwise get swept mid-flight,
|
|
# racing phase3's commit. Audit 2026-06-02.
|
|
existing.started_at = datetime.now(UTC)
|
|
await self.async_session.commit()
|
|
event_id = existing.id
|
|
else:
|
|
ev = DownloadEvent(source_id=source_id, status="running")
|
|
self.async_session.add(ev)
|
|
await self.async_session.commit()
|
|
await self.async_session.refresh(ev)
|
|
event_id = ev.id
|
|
|
|
artist = source.artist
|
|
# Drive cookies-vs-token selection from the platform registry's
|
|
# auth_type so a new 7th token-platform automatically picks the
|
|
# right credential path. The hardcoded tuple here used to drift
|
|
# out of sync with credential_service's auth_type_for(). Audit
|
|
# 2026-06-02.
|
|
if auth_type_for(source.platform) == "token":
|
|
cookies_path = None
|
|
auth_token = await self.cred_service.get_token(source.platform)
|
|
else:
|
|
cookies_path_obj = await self.cred_service.get_cookies_path(source.platform)
|
|
cookies_path = str(cookies_path_obj) if cookies_path_obj else None
|
|
auth_token = None
|
|
|
|
return {
|
|
"status": "ok",
|
|
"event_id": event_id,
|
|
"source_id": source_id,
|
|
"url": source.url,
|
|
"platform": source.platform,
|
|
"artist_slug": artist.slug if artist else "_unknown",
|
|
"artist_id": artist.id if artist else None,
|
|
"config_overrides": dict(source.config_overrides or {}),
|
|
"cookies_path": cookies_path,
|
|
"auth_token": auth_token,
|
|
"backfill_runs_remaining": source.backfill_runs_remaining or 0,
|
|
}
|
|
|
|
async def _phase3_persist(
|
|
self,
|
|
event_id: int,
|
|
ctx: dict,
|
|
dl_result,
|
|
resolved_campaign_id: str | None,
|
|
) -> int:
|
|
# Cache Patreon campaign id on the source.
|
|
if resolved_campaign_id:
|
|
src = self.sync_session.get(Source, ctx["source_id"])
|
|
if src is not None:
|
|
src.config_overrides = {
|
|
**(src.config_overrides or {}),
|
|
"patreon_campaign_id": resolved_campaign_id,
|
|
}
|
|
self.sync_session.commit()
|
|
|
|
artist = None
|
|
source_row = None
|
|
if ctx.get("artist_id"):
|
|
artist = self.sync_session.get(Artist, ctx["artist_id"])
|
|
source_row = self.sync_session.get(Source, ctx["source_id"])
|
|
|
|
import_summary = {"attached": 0, "skipped": 0, "errors": 0}
|
|
bytes_downloaded = 0
|
|
|
|
loop = asyncio.get_running_loop()
|
|
|
|
for path_str in dl_result.written_paths or []:
|
|
if path_str in (dl_result.quarantined_paths or []):
|
|
continue
|
|
path = Path(path_str)
|
|
# Sync stdlib filesystem ops are intentional here: this orchestrator
|
|
# runs under `asyncio.run` inside a Celery task — no other awaitables
|
|
# compete for the loop. ASYNC240 suppressed below.
|
|
if not path.exists(): # noqa: ASYNC240
|
|
continue
|
|
|
|
def _attach(p=path):
|
|
return self.importer.attach_in_place(
|
|
p, artist=artist, source=source_row,
|
|
)
|
|
|
|
result = await loop.run_in_executor(None, _attach)
|
|
|
|
if result.status in ("imported", "superseded"):
|
|
import_summary["attached"] += 1
|
|
try:
|
|
bytes_downloaded += path.stat().st_size # noqa: ASYNC240
|
|
except OSError:
|
|
pass
|
|
# Enqueue thumbnail + ML for newly-attached images, matching
|
|
# the filesystem-import path (tasks/import_file.py:228-239).
|
|
# Importer.attach_in_place deliberately skips inline thumb
|
|
# generation to keep the import queue moving; the calling
|
|
# task is responsible for the enqueue. Operator-flagged
|
|
# 2026-06-01: without this, every downloaded image stayed
|
|
# at thumbnail_path=NULL until a periodic backfill swept
|
|
# it up, surfacing as broken-thumbnail tiles in the gallery
|
|
# for hours after a download landed. Lazy import to avoid
|
|
# circular-import risk between this service and the
|
|
# tasks/* modules that import it.
|
|
from ..tasks.ml import tag_and_embed
|
|
from ..tasks.thumbnail import generate_thumbnail
|
|
ids = list(result.member_image_ids)
|
|
if result.image_id is not None and result.image_id not in ids:
|
|
ids.append(result.image_id)
|
|
for img_id in ids:
|
|
generate_thumbnail.delay(img_id)
|
|
tag_and_embed.delay(img_id)
|
|
elif result.status == "attached":
|
|
# Non-media or extracted archive captured as PostAttachment
|
|
# (FC-2d-iii). The canonical copy lives in the attachments
|
|
# store; the original download path is now redundant —
|
|
# mirror duplicate_hash cleanup so we don't keep two copies.
|
|
# Operator-flagged 2026-06-02 (Lustria OST zip).
|
|
import_summary["attached"] += 1
|
|
try:
|
|
bytes_downloaded += path.stat().st_size # noqa: ASYNC240
|
|
except OSError:
|
|
pass
|
|
try:
|
|
path.unlink(missing_ok=True) # noqa: ASYNC240
|
|
except OSError:
|
|
pass
|
|
elif result.status == "skipped" and result.skip_reason and result.skip_reason.value in (
|
|
"duplicate_hash", "duplicate_phash",
|
|
):
|
|
import_summary["skipped"] += 1
|
|
try:
|
|
path.unlink(missing_ok=True) # noqa: ASYNC240
|
|
except OSError:
|
|
pass
|
|
elif result.status == "skipped":
|
|
# Soft skip (too_small, too_transparent, invalid_image) —
|
|
# the file just didn't qualify, not a download/ingest
|
|
# 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
|
|
|
|
ev = (await self.async_session.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
|
|
run_stats = self.gdl._compute_run_stats(
|
|
dl_result.return_code, dl_result.stdout, dl_result.stderr
|
|
)
|
|
run_stats["quarantined_count"] = dl_result.files_quarantined
|
|
stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr)
|
|
|
|
# Plan #544: PARTIAL means the run downloaded ≥1 file but the
|
|
# subprocess didn't finish in budget (typically wall-clock timeout
|
|
# mid-walk). Real work happened; the next tick continues via
|
|
# gallery-dl's archive. NOT a failure for status purposes.
|
|
if dl_result.success and import_summary["errors"] == 0:
|
|
status = "ok"
|
|
elif dl_result.error_type == ErrorType.PARTIAL and import_summary["errors"] == 0:
|
|
status = "ok"
|
|
else:
|
|
status = "error"
|
|
ev.status = status
|
|
ev.finished_at = datetime.now(UTC)
|
|
ev.files_count = import_summary["attached"]
|
|
ev.bytes_downloaded = bytes_downloaded
|
|
ev.error = dl_result.error_message if status == "error" else None
|
|
ev.metadata_ = {
|
|
"run_stats": run_stats,
|
|
"error_type": dl_result.error_type.value if dl_result.error_type else None,
|
|
"stdout": self.gdl._truncate_log(dl_result.stdout) or None,
|
|
"stderr": self.gdl._truncate_log(dl_result.stderr) or None,
|
|
"stderr_errors_warnings": stderr_summary or None,
|
|
"duration_seconds": dl_result.duration_seconds,
|
|
"quarantined_paths": dl_result.quarantined_paths or None,
|
|
"import_summary": import_summary,
|
|
}
|
|
await self._update_source_health(
|
|
source_id=ctx["source_id"], status=status, error_message=ev.error,
|
|
error_type=dl_result.error_type.value if dl_result.error_type else None,
|
|
)
|
|
await self._apply_backfill_lifecycle(ctx, dl_result)
|
|
await self.async_session.commit()
|
|
return event_id
|
|
|
|
async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None:
|
|
"""Cursor-paged backfill state machine (plan #689).
|
|
|
|
A backfill is "in progress" while backfill_runs_remaining > 0 OR a
|
|
checkpoint cursor is stored. Completion is a CLEAN rc=0 finish —
|
|
gallery-dl with skip:True exits 0 only after exhausting the
|
|
newest→oldest walk; a run cut short by the subprocess budget returns
|
|
success=False / return_code=-1 (the TimeoutExpired path), so rc=0 is
|
|
an unambiguous "reached the bottom" signal regardless of how many
|
|
files this run fetched. (This replaces plan #544's files==0 predicate,
|
|
whose audit-2026-06-02 VALIDATION_FAILED guard still holds: those exit
|
|
non-zero or carry an error_type, so they are not "completed".)
|
|
|
|
For patreon, each non-completing run checkpoints gallery-dl's
|
|
last-emitted pagination cursor — even a timed-out run, since its
|
|
partial stdout/stderr still carries it — so the next run RESUMES
|
|
instead of re-walking from the top. A stuck-guard clears the cursor
|
|
after two consecutive runs that fail to advance it, so a wedged walk
|
|
can't re-strand forever (cf. download soft-limit ladder).
|
|
"""
|
|
overrides = ctx["config_overrides"] or {}
|
|
old_cursor = overrides.get("_backfill_cursor")
|
|
backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0
|
|
if backfill_remaining <= 0 and not old_cursor:
|
|
return # tick mode — counter untouched
|
|
|
|
src = (await self.async_session.execute(
|
|
select(Source).where(Source.id == ctx["source_id"])
|
|
)).scalar_one()
|
|
new_overrides = dict(src.config_overrides or {})
|
|
|
|
completed = (
|
|
dl_result.success
|
|
and dl_result.error_type is None
|
|
and dl_result.return_code == 0
|
|
)
|
|
if completed:
|
|
new_overrides.pop("_backfill_cursor", None)
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
src.config_overrides = new_overrides
|
|
src.backfill_runs_remaining = 0
|
|
return
|
|
|
|
# Non-completing run. Patreon checkpoints + resumes via cursor;
|
|
# other platforms have no resumable cursor, so just decrement.
|
|
if ctx["platform"] == "patreon":
|
|
new_cursor = parse_last_cursor(dl_result.stdout, dl_result.stderr)
|
|
if new_cursor and new_cursor != old_cursor:
|
|
new_overrides["_backfill_cursor"] = new_cursor
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
else:
|
|
stalls = int(new_overrides.get("_backfill_cursor_stalls", 0)) + 1
|
|
if stalls >= 2:
|
|
# Wedged: give up, drop back to tick mode. The surfaced
|
|
# error_type on the event tells the operator why.
|
|
new_overrides.pop("_backfill_cursor", None)
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
src.config_overrides = new_overrides
|
|
src.backfill_runs_remaining = 0
|
|
return
|
|
new_overrides["_backfill_cursor_stalls"] = stalls
|
|
src.config_overrides = new_overrides
|
|
|
|
src.backfill_runs_remaining = max(0, backfill_remaining - 1)
|
|
|
|
async def _update_source_health(
|
|
self, *, source_id: int, status: str, error_message: str | None,
|
|
error_type: str | None = None,
|
|
) -> None:
|
|
"""FC-3d: update Source.{consecutive_failures, last_error, last_checked_at}.
|
|
|
|
ok -> failures = 0, error = None, checked_at = now
|
|
error -> failures += 1, error = error_message, checked_at = now
|
|
skipped -> failures unchanged, error = None, checked_at = now
|
|
|
|
When error_type == 'rate_limited', also stamps a platform-wide
|
|
cooldown via scheduler_service.set_platform_cooldown so the next
|
|
scan tick skips every source on this platform until the cooldown
|
|
expires. Preventive half of the burst-prevention pair —
|
|
consecutive_failures still backs the offending source off across
|
|
ticks.
|
|
"""
|
|
source = (await self.async_session.execute(
|
|
select(Source).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
now = datetime.now(UTC)
|
|
if status == "ok":
|
|
source.consecutive_failures = 0
|
|
source.last_error = None
|
|
# alembic 0032 — clear the failure-class chip on success.
|
|
source.error_type = None
|
|
elif status == "error":
|
|
source.consecutive_failures = (source.consecutive_failures or 0) + 1
|
|
source.last_error = error_message
|
|
# alembic 0032 — stamp the failure-class so FailingSourcesCard
|
|
# can render a colored chip and operators can bulk-triage
|
|
# by error class without opening Logs per row.
|
|
source.error_type = error_type
|
|
if error_type == "rate_limited":
|
|
await set_platform_cooldown(self.async_session, source.platform)
|
|
elif status == "skipped":
|
|
source.last_error = None
|
|
source.last_checked_at = now
|
|
|
|
async def _finalize_event_and_source(
|
|
self, *, event_id: int, source_id: int, status: str,
|
|
error_message: str | None,
|
|
) -> None:
|
|
"""Finalize the event AND the source-health columns in one
|
|
transactional step. Used by the skipped early-out path and by
|
|
unit tests that exercise the source-health writes directly.
|
|
"""
|
|
ev = (await self.async_session.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
ev.status = status
|
|
ev.finished_at = datetime.now(UTC)
|
|
ev.error = error_message
|
|
await self._update_source_health(
|
|
source_id=source_id, status=status, error_message=error_message,
|
|
)
|
|
await self.async_session.commit()
|