bvandeusen
19eb4e9388
The recurring "post shows a zip but no images" report had no diagnostic: when
_import_archive captured an archive as a bare PostAttachment — because the
bomb-guard probe rejected it, or extraction yielded zero members (corrupt /
unsupported / missing extractor backend), or it held only non-media files — it
returned status="attached" silently.
Now those paths set ImportResult.error with the specific reason and log a
warning, and download_service records each as {file, reason} under the event's
metadata.unextracted_archives (None when every archive extracted cleanly). So
the next run names exactly which archives failed and why, instead of leaving the
operator to guess. No behaviour change to the happy path.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
590 lines
28 KiB
Python
590 lines
28 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
|
|
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 .download_backends import run_download, uses_native_ingester
|
|
from .gallery_dl import (
|
|
BACKFILL_CHUNK_SECONDS,
|
|
BACKFILL_SKIP_VALUE,
|
|
TICK_SKIP_VALUE,
|
|
DownloadResult,
|
|
ErrorType,
|
|
GalleryDLService,
|
|
SourceConfig,
|
|
extract_errors_warnings,
|
|
truncate_log,
|
|
)
|
|
from .importer import Importer
|
|
from .platforms import auth_type_for
|
|
from .scheduler_service import set_platform_cooldown
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
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,
|
|
sync_session_factory=None,
|
|
):
|
|
self.async_session = async_session
|
|
self.sync_session = sync_session
|
|
self.gdl = gdl
|
|
self.importer = importer
|
|
self.cred_service = cred_service
|
|
# Sync sessionmaker the native Patreon ingester opens SHORT-LIVED
|
|
# sessions from for its seen-ledger reads/writes (never one held across
|
|
# the multi-minute walk — see PatreonIngester). Only the patreon branch
|
|
# of phase 2 uses it; gallery-dl sources leave it None.
|
|
self.sync_session_factory = sync_session_factory
|
|
|
|
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 {})
|
|
# Backfill mode (plan #693): the source's `_backfill_state == "running"`
|
|
# selects a time-boxed deep-walk chunk — skip: True (walk full history)
|
|
# + the BACKFILL_CHUNK_SECONDS budget, resuming from the cursor
|
|
# checkpoint (plan #689). State is the single source of truth; it stays
|
|
# "running" across ticks until the walk reaches the bottom (phase 3
|
|
# flips it to "complete"). Otherwise tick mode: exit gallery-dl after
|
|
# 20 contiguous archived items (skip: "exit:20" + the default 870s).
|
|
# Operator drives this via POST /api/sources/{id}/backfill {action}.
|
|
overrides = ctx["config_overrides"] or {}
|
|
in_backfill = overrides.get("_backfill_state") == "running"
|
|
# Recovery (plan #697) reuses the entire #693 backfill state machine —
|
|
# cursor checkpoint, time-boxed chunks, complete/stall lifecycle — and
|
|
# differs only in bypassing the tier-1 seen-ledger so dropped-and-deleted
|
|
# near-dups get re-fetched and re-evaluated under the CURRENT pHash
|
|
# threshold (tier-2 disk still spares files we kept). The
|
|
# `_backfill_bypass_seen` flag rides alongside the running backfill state;
|
|
# download mode is "recovery" when both are set.
|
|
bypass_seen = bool(overrides.get("_backfill_bypass_seen"))
|
|
if in_backfill:
|
|
skip_value: bool | str = BACKFILL_SKIP_VALUE
|
|
source_config.timeout = BACKFILL_CHUNK_SECONDS
|
|
pending_cursor = overrides.get("_backfill_cursor")
|
|
if uses_native_ingester(ctx["platform"]) and pending_cursor:
|
|
source_config.resume_cursor = pending_cursor
|
|
else:
|
|
skip_value = TICK_SKIP_VALUE
|
|
|
|
# Phase 2 dispatch is uniform across backends (download_backends.
|
|
# run_download — the download counterpart to verify_source_credential):
|
|
# native platforms run their ingester in `mode` (zero per-file HEADs,
|
|
# native cursor/resume, loud drift detection); gallery-dl platforms run
|
|
# the subprocess. Either returns a DownloadResult-shaped object so phase 3
|
|
# is untouched. `mode` is None for gallery-dl (run_download ignores it).
|
|
mode: str | None = None
|
|
if uses_native_ingester(ctx["platform"]):
|
|
if in_backfill and bypass_seen:
|
|
mode = "recovery"
|
|
elif in_backfill:
|
|
mode = "backfill"
|
|
else:
|
|
mode = "tick"
|
|
dl_result, resolved_campaign_id = await self._run_download(
|
|
ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
|
|
)
|
|
|
|
# A backfill chunk that hit its time-box but made forward progress is
|
|
# NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as
|
|
# "ok/progress" (PARTIAL maps to status "ok"), not a red error, and the
|
|
# next chunk just resumes from the new cursor (plan #693). A chunk that
|
|
# timed out with NO progress stays TIMEOUT and feeds phase 3's
|
|
# stall-guard. Leave RATE_LIMITED alone so the platform-cooldown fires.
|
|
if in_backfill and dl_result.error_type == ErrorType.TIMEOUT:
|
|
new_cursor = dl_result.cursor # plan #704: structured, not scraped
|
|
advanced = bool(
|
|
(new_cursor and new_cursor != overrides.get("_backfill_cursor"))
|
|
or dl_result.files_downloaded > 0
|
|
)
|
|
if advanced:
|
|
dl_result.error_type = ErrorType.PARTIAL
|
|
dl_result.error_message = (
|
|
f"Backfill chunk: {dl_result.files_downloaded} file(s) — continuing"
|
|
)
|
|
|
|
return await self._phase3_persist(
|
|
ctx["event_id"], ctx, dl_result, resolved_campaign_id,
|
|
)
|
|
|
|
async def _run_download(
|
|
self, *, ctx: dict, source_config, skip_value, mode: str | None,
|
|
) -> tuple[DownloadResult, str | None]:
|
|
"""Phase-2 dispatch → `download_backends.run_download`, passing this
|
|
service's gdl + sync sessionmaker. Kept as a thin instance method so tests
|
|
can stub the whole phase-2 dispatch on the service, and so the per-backend
|
|
construction lives in download_backends (the backend registry)."""
|
|
return await run_download(
|
|
ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
|
|
gdl=self.gdl, sync_session_factory=self.sync_session_factory,
|
|
)
|
|
|
|
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}
|
|
# Archives detected but captured WITHOUT extracting any image (probe
|
|
# rejected / corrupt / missing extractor backend). Surfaced on the event
|
|
# so a post showing "no images" beside a zip is diagnosable (plan
|
|
# follow-up 2026-06-06 — the recurring archive-association report).
|
|
unextracted_archives: list[dict] = []
|
|
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
|
|
# An archive captured WITHOUT extracting any image carries a
|
|
# reason on result.error — record it so the event explains the
|
|
# "no images beside a zip" symptom instead of staying silent.
|
|
if result.error:
|
|
unextracted_archives.append(
|
|
{"file": path.name, "reason": result.error}
|
|
)
|
|
log.warning(
|
|
"archive captured unextracted (%s): %s",
|
|
path.name, result.error,
|
|
)
|
|
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()
|
|
|
|
# plan #704: the native ingester returns structured run_stats; only the
|
|
# gallery-dl path needs the regex-over-stdout reconstruction.
|
|
if dl_result.run_stats is not None:
|
|
run_stats = dict(dl_result.run_stats)
|
|
else:
|
|
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 = 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": truncate_log(dl_result.stdout) or None,
|
|
"stderr": 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,
|
|
# Archives detected but captured without extracting an image — the
|
|
# recurring "post shows a zip but no images" report. Each entry is
|
|
# {file, reason}; None when every archive extracted cleanly.
|
|
"unextracted_archives": unextracted_archives or None,
|
|
}
|
|
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,
|
|
retry_after_seconds=getattr(dl_result, "retry_after_seconds", 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:
|
|
"""Backfill state machine (plan #693, building on the cursor of #689).
|
|
|
|
A backfill runs in time-boxed chunks while
|
|
`config_overrides["_backfill_state"] == "running"`. Each chunk:
|
|
- COMPLETES the walk → clean rc=0 (gallery-dl with skip:True exits 0
|
|
only after exhausting the newest→oldest walk; a chunk cut short by
|
|
its time-box returns success=False / rc<0 via TimeoutExpired). On
|
|
completion: state="complete", clear the cursor, return to tick mode.
|
|
- made PROGRESS (cursor advanced and/or files written) → stay
|
|
"running", checkpoint the new cursor, bump the chunk counter, and
|
|
spend one of the safety-cap chunks (backfill_runs_remaining). If the
|
|
cap is exhausted without finishing → state="stalled".
|
|
- made NO progress → increment the stall counter; two strikes →
|
|
state="stalled", clear the cursor (a wedged walk can't loop).
|
|
|
|
State is the single source of truth; the cursor lives only inside a
|
|
running backfill. config_overrides is reassigned (not mutated in place)
|
|
for SQLAlchemy JSON change detection.
|
|
"""
|
|
overrides = ctx["config_overrides"] or {}
|
|
if overrides.get("_backfill_state") != "running":
|
|
return # not backfilling — tick mode, nothing to do
|
|
|
|
old_cursor = overrides.get("_backfill_cursor")
|
|
cap_remaining = ctx.get("backfill_runs_remaining", 0) or 0
|
|
|
|
src = (await self.async_session.execute(
|
|
select(Source).where(Source.id == ctx["source_id"])
|
|
)).scalar_one()
|
|
new_overrides = dict(src.config_overrides or {})
|
|
chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
|
|
new_overrides["_backfill_chunks"] = chunks
|
|
# plan #704 (#5): _backfill_posts (the live progress badge) is OWNED by the
|
|
# ingester now — it writes a monotonic absolute mid-walk at each page
|
|
# boundary (ingest_core._checkpoint_posts), so the badge climbs DURING a
|
|
# chunk instead of jumping once per chunk here, and the re-walked resume
|
|
# page is no longer double-counted. new_overrides (read fresh above)
|
|
# carries the ingester's committed value forward untouched.
|
|
|
|
completed = (
|
|
dl_result.success
|
|
and dl_result.error_type is None
|
|
and dl_result.return_code == 0
|
|
)
|
|
if completed:
|
|
new_overrides["_backfill_state"] = "complete"
|
|
new_overrides.pop("_backfill_cursor", None)
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
# plan #697: a recovery walk shares this lifecycle; clear its bypass
|
|
# flag on completion so the next routine tick honors the seen-ledger.
|
|
new_overrides.pop("_backfill_bypass_seen", None)
|
|
src.config_overrides = new_overrides
|
|
src.backfill_runs_remaining = 0
|
|
return
|
|
|
|
# Did not finish. The native ingester checkpoints + resumes via cursor
|
|
# (carried structurally on the result, plan #704); gallery-dl platforms
|
|
# have no resumable cursor (every chunk re-walks from the top), so they
|
|
# advance only by the download archive growing.
|
|
new_cursor = (
|
|
dl_result.cursor if uses_native_ingester(ctx["platform"]) else None
|
|
)
|
|
advanced = bool(
|
|
(new_cursor and new_cursor != old_cursor)
|
|
or dl_result.files_downloaded > 0
|
|
)
|
|
if advanced:
|
|
if new_cursor:
|
|
new_overrides["_backfill_cursor"] = new_cursor
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
cap_remaining = max(0, cap_remaining - 1)
|
|
src.backfill_runs_remaining = cap_remaining
|
|
if cap_remaining == 0:
|
|
# Safety cap hit before reaching the bottom — pause, don't loop.
|
|
new_overrides["_backfill_state"] = "stalled"
|
|
else:
|
|
stalls = int(new_overrides.get("_backfill_cursor_stalls", 0)) + 1
|
|
if stalls >= 2:
|
|
new_overrides["_backfill_state"] = "stalled"
|
|
new_overrides.pop("_backfill_cursor", None)
|
|
new_overrides.pop("_backfill_cursor_stalls", None)
|
|
else:
|
|
new_overrides["_backfill_cursor_stalls"] = stalls
|
|
|
|
src.config_overrides = new_overrides
|
|
|
|
async def _update_source_health(
|
|
self, *, source_id: int, status: str, error_message: str | None,
|
|
error_type: str | None = None, retry_after_seconds: float | 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":
|
|
# plan #708 B1: honor the server's Retry-After when the native
|
|
# client surfaced one (clamped to a sane [60, 3600] window so a
|
|
# tiny hint can't leave the platform effectively un-cooled and a
|
|
# huge one can't strand it for hours); else the flat default.
|
|
if retry_after_seconds is not None:
|
|
seconds = int(min(max(retry_after_seconds, 60), 3600))
|
|
await set_platform_cooldown(
|
|
self.async_session, source.platform, seconds=seconds,
|
|
)
|
|
else:
|
|
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()
|