Surgical re-fetch for deep posts (ExternalLink reset) — recovery no longer needs deep scans #189
@@ -7,6 +7,7 @@ Action surfaces:
|
||||
POST /api/admin/tags/<int:dest_id>/merge (Tier B)
|
||||
POST /api/admin/tags/prune-unused (Tier A)
|
||||
POST /api/admin/posts/prune-bare (Tier A)
|
||||
POST /api/admin/posts/refetch-external (Tier A)
|
||||
GET /api/admin/tags/<int:tag_id>/usage-count (helper)
|
||||
|
||||
Tier-C ops take a dry_run body flag (returns projection inline,
|
||||
@@ -22,7 +23,7 @@ from quart import Blueprint, jsonify, request
|
||||
from sqlalchemy import select, text
|
||||
|
||||
from ..extensions import get_session
|
||||
from ..models import Artist
|
||||
from ..models import Artist, Post
|
||||
from ..services.cleanup_service import project_artist_cascade, project_bulk_image_delete
|
||||
from ._responses import error_response as _bad
|
||||
|
||||
@@ -276,6 +277,44 @@ async def posts_reconcile_duplicates():
|
||||
return await _run_dry_run_op(reconcile_duplicate_posts, source_id=source_id)
|
||||
|
||||
|
||||
@admin_bp.route("/posts/refetch-external", methods=["POST"])
|
||||
async def posts_refetch_external():
|
||||
"""Surgical re-fetch of a post's external file-host links (operator
|
||||
2026-07-03): the normal cadence never re-walks deep back-catalogue posts,
|
||||
so a deleted external file only comes back by resetting its ExternalLink
|
||||
row(s) — this endpoint does that per post and dispatches the fetches.
|
||||
Sha-dedupe discards payload files that still exist, so only what's
|
||||
missing lands. Body: {external_post_id: str, source_id?: int (to
|
||||
disambiguate the same external id across sources)}."""
|
||||
from ..services.external_links import refetch_links_for_post
|
||||
|
||||
body = await request.get_json(silent=True) or {}
|
||||
ext_id = str(body.get("external_post_id") or "").strip()
|
||||
if not ext_id:
|
||||
return _bad("missing_external_post_id",
|
||||
detail="external_post_id is required")
|
||||
raw_source = body.get("source_id")
|
||||
try:
|
||||
source_id = int(raw_source) if raw_source is not None else None
|
||||
except (TypeError, ValueError):
|
||||
return _bad("invalid_source_id", detail="source_id must be an integer")
|
||||
|
||||
async with get_session() as session:
|
||||
stmt = select(Post.id).where(Post.external_post_id == ext_id)
|
||||
if source_id is not None:
|
||||
stmt = stmt.where(Post.source_id == source_id)
|
||||
post_ids = (await session.execute(stmt)).scalars().all()
|
||||
if not post_ids:
|
||||
return _bad("post_not_found", status=404,
|
||||
detail=f"no post with external_post_id {ext_id!r}")
|
||||
results = {}
|
||||
for pid in post_ids:
|
||||
results[str(pid)] = await session.run_sync(
|
||||
lambda s, p=pid: refetch_links_for_post(s, p)
|
||||
)
|
||||
return jsonify({"posts": results})
|
||||
|
||||
|
||||
def _reset_content_confirm_token(projection: dict) -> str:
|
||||
"""Stable 8-hex token derived from the live counts (mirrors the Tier-C
|
||||
bulk-delete token): it changes whenever the data changes, so the apply can
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Surgical re-fetch of a post's external file-host links.
|
||||
|
||||
The normal download cadence never re-walks deep back-catalogue posts (the
|
||||
seen-gates exist precisely to keep old items from resurfacing), so when a
|
||||
file that CAME from an ExternalLink is deleted — e.g. the failure-triage
|
||||
recovery flow removing a corrupt original — a plain source re-check will
|
||||
never bring it back. The link ROW is the durable, per-post handle: resetting
|
||||
it to pending and dispatching the fetch re-downloads exactly that link's
|
||||
payload, and sha-dedupe at import discards anything that still exists — so
|
||||
only the missing file actually lands. (Operator 2026-07-03: recovery must
|
||||
not require artist-wide deep scans.)
|
||||
"""
|
||||
import logging
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..models import ExternalLink
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def refetch_links_for_post(session: Session, post_id: int) -> dict:
|
||||
"""Reset every settled ExternalLink on a post to pending (fresh attempt
|
||||
budget) and dispatch its fetch. In-flight ('downloading') links are left
|
||||
alone. Commits. Returns {links_total, links_reset}."""
|
||||
links = session.execute(
|
||||
select(ExternalLink).where(ExternalLink.post_id == post_id)
|
||||
).scalars().all()
|
||||
reset_ids = []
|
||||
for link in links:
|
||||
if link.status == "downloading":
|
||||
continue
|
||||
link.status = "pending"
|
||||
link.attempts = 0
|
||||
link.last_error = None
|
||||
link.completed_at = None
|
||||
reset_ids.append(link.id)
|
||||
session.commit()
|
||||
if reset_ids:
|
||||
# Lazy import (services -> tasks would cycle at module load). The
|
||||
# 10-min extdl sweep would pick pending rows up anyway — dispatching
|
||||
# directly just skips the wait.
|
||||
from ..tasks.external import fetch_external_link
|
||||
|
||||
for lid in reset_ids:
|
||||
fetch_external_link.delay(lid)
|
||||
log.info(
|
||||
"external refetch: post %s — %d/%d link(s) reset + dispatched",
|
||||
post_id, len(reset_ids), len(links),
|
||||
)
|
||||
return {"links_total": len(links), "links_reset": len(reset_ids)}
|
||||
@@ -121,13 +121,22 @@ def triage_errored_jobs(
|
||||
def recover_defective_image(
|
||||
session: Session, image_id: int, *, images_root: Path,
|
||||
) -> dict:
|
||||
"""Delete the defective copy + record and re-poll its subscription Source.
|
||||
"""Delete the defective copy + record and queue its re-fetch — surgically
|
||||
where possible.
|
||||
|
||||
Mirrors the Layer-2 import refetch: with the bad file gone, the source's
|
||||
next gallery-dl run re-fetches a fresh copy, which re-imports as a new
|
||||
record and re-enters the GPU pipeline. The record delete cascades the
|
||||
error tombstones with it. 'no_source' when no enabled, real-URL Source is
|
||||
reachable via the image's provenance — manual remediation there."""
|
||||
Two re-fetch layers (operator 2026-07-03: deep back-catalogue items are
|
||||
NEVER re-walked by the normal cadence, so recovery can't rely on it):
|
||||
1. SURGICAL: any ExternalLink rows on the image's post(s) are reset +
|
||||
re-dispatched — this is how external-host files (the common defect
|
||||
case: big videos) come back regardless of post age. Sha-dedupe at
|
||||
import discards payload files that still exist.
|
||||
2. BROAD: a source re-check, which re-fetches gallery-dl-NATIVE files the
|
||||
walk still reaches (recent posts). A native file deeper than the walk
|
||||
needs a per-source backfill/deep scan — reported via links_reset=0 so
|
||||
the caller can say so.
|
||||
|
||||
The record delete cascades the error tombstones with it. 'no_source' when
|
||||
no enabled, real-URL Source resolves via provenance — manual there."""
|
||||
rec = session.get(ImageRecord, image_id)
|
||||
if rec is None:
|
||||
return {"status": "not_found"}
|
||||
@@ -143,14 +152,28 @@ def recover_defective_image(
|
||||
).scalars().first()
|
||||
if src_id is None:
|
||||
return {"status": "no_source"}
|
||||
# Capture the post linkage BEFORE the delete cascades provenance away.
|
||||
post_ids = session.execute(
|
||||
select(ImageProvenance.post_id)
|
||||
.where(ImageProvenance.image_record_id == image_id)
|
||||
).scalars().all()
|
||||
path = rec.path
|
||||
summary = delete_images(session, image_ids=[image_id], images_root=images_root)
|
||||
from ..external_links import refetch_links_for_post
|
||||
|
||||
links_reset = 0
|
||||
for pid in post_ids:
|
||||
links_reset += refetch_links_for_post(session, pid)["links_reset"]
|
||||
# Lazy import (services -> tasks would cycle at module load).
|
||||
from ...tasks.download import download_source
|
||||
|
||||
download_source.delay(src_id)
|
||||
log.warning(
|
||||
"gpu triage recovery: deleted defective image %s (%s) and queued a "
|
||||
"re-check of source %s to re-fetch it", image_id, path, src_id,
|
||||
"gpu triage recovery: deleted defective image %s (%s); reset %d "
|
||||
"external link(s) and queued a re-check of source %s",
|
||||
image_id, path, links_reset, src_id,
|
||||
)
|
||||
return {"status": "refetch_queued", "source_id": src_id, **summary}
|
||||
return {
|
||||
"status": "refetch_queued", "source_id": src_id,
|
||||
"links_reset": links_reset, **summary,
|
||||
}
|
||||
|
||||
@@ -390,6 +390,72 @@ async def test_prune_unused_commit_deletes_and_returns_count(client, db):
|
||||
assert "byebye" in body["sample_names"]
|
||||
|
||||
|
||||
# --- Tier-A: POST /posts/refetch-external ----------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refetch_external_resets_and_dispatches(client, db, monkeypatch):
|
||||
"""Surgical recovery for deep posts (2026-07-03): settled links reset to
|
||||
pending with a fresh budget + dispatched; in-flight links untouched."""
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.models import ExternalLink, Post, Source
|
||||
|
||||
a = Artist(name="Deep", slug="deep")
|
||||
db.add(a)
|
||||
await db.flush()
|
||||
src = Source(artist_id=a.id, platform="patreon",
|
||||
url="https://www.patreon.com/deep", enabled=True)
|
||||
db.add(src)
|
||||
await db.flush()
|
||||
post = Post(artist_id=a.id, source_id=src.id, external_post_id="96376640")
|
||||
db.add(post)
|
||||
await db.flush()
|
||||
done = ExternalLink(post_id=post.id, artist_id=a.id, host="mega",
|
||||
url="https://mega.nz/file/a#k1", status="downloaded",
|
||||
attempts=1)
|
||||
dead = ExternalLink(post_id=post.id, artist_id=a.id, host="gdrive",
|
||||
url="https://drive.google.com/x", status="dead",
|
||||
attempts=5, last_error="quota")
|
||||
busy = ExternalLink(post_id=post.id, artist_id=a.id, host="mega",
|
||||
url="https://mega.nz/file/b#k2", status="downloading",
|
||||
attempts=1)
|
||||
db.add_all([done, dead, busy])
|
||||
await db.flush()
|
||||
ids = (done.id, dead.id, busy.id)
|
||||
await db.commit()
|
||||
|
||||
fetches = []
|
||||
monkeypatch.setattr(
|
||||
"backend.app.tasks.external.fetch_external_link.delay",
|
||||
lambda lid: fetches.append(lid),
|
||||
)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/admin/posts/refetch-external",
|
||||
json={"external_post_id": "96376640"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = await resp.get_json()
|
||||
assert list(body["posts"].values()) == [{"links_total": 3, "links_reset": 2}]
|
||||
assert sorted(fetches) == sorted([ids[0], ids[1]])
|
||||
|
||||
rows = dict((await db.execute(
|
||||
select(ExternalLink.id, ExternalLink.status)
|
||||
.where(ExternalLink.post_id == post.id)
|
||||
)).all())
|
||||
assert rows[ids[0]] == "pending"
|
||||
assert rows[ids[1]] == "pending"
|
||||
assert rows[ids[2]] == "downloading" # in-flight left alone
|
||||
|
||||
# Unknown post → 404, nothing dispatched.
|
||||
miss = await client.post(
|
||||
"/api/admin/posts/refetch-external",
|
||||
json={"external_post_id": "nope"},
|
||||
)
|
||||
assert miss.status_code == 404
|
||||
|
||||
|
||||
# --- Tier-A: POST /posts/prune-bare + /posts/reconcile-duplicates ---
|
||||
# These two routes share _run_dry_run_op with the tag prunes (DRY pass,
|
||||
# task #753); cover their apply path + reconcile's source_id passthrough so
|
||||
|
||||
@@ -12,6 +12,7 @@ from sqlalchemy import select
|
||||
|
||||
from backend.app.models import (
|
||||
Artist,
|
||||
ExternalLink,
|
||||
GpuJob,
|
||||
ImageProvenance,
|
||||
ImageRecord,
|
||||
@@ -138,7 +139,14 @@ async def test_recover_deletes_record_and_requeues_source(
|
||||
await db.flush()
|
||||
db.add(ImageProvenance(image_record_id=img.id, post_id=post.id,
|
||||
source_id=src.id))
|
||||
img_id, src_id = img.id, src.id
|
||||
# A settled external-host link on the same post: recovery must reset it
|
||||
# (the SURGICAL path — deep posts are never re-walked by the cadence).
|
||||
link = ExternalLink(post_id=post.id, artist_id=artist.id, host="mega",
|
||||
url="https://mega.nz/file/x#key", status="downloaded",
|
||||
attempts=2)
|
||||
db.add(link)
|
||||
await db.flush()
|
||||
img_id, src_id, link_id = img.id, src.id, link.id
|
||||
await db.commit()
|
||||
|
||||
queued = []
|
||||
@@ -146,13 +154,27 @@ async def test_recover_deletes_record_and_requeues_source(
|
||||
"backend.app.tasks.download.download_source.delay",
|
||||
lambda sid: queued.append(sid),
|
||||
)
|
||||
fetches = []
|
||||
monkeypatch.setattr(
|
||||
"backend.app.tasks.external.fetch_external_link.delay",
|
||||
lambda lid: fetches.append(lid),
|
||||
)
|
||||
|
||||
resp = await client.post(f"/api/gpu/errors/{img_id}/recover")
|
||||
assert resp.status_code == 200
|
||||
body = await resp.get_json()
|
||||
assert body["status"] == "refetch_queued"
|
||||
assert body["source_id"] == src_id
|
||||
assert body["links_reset"] == 1
|
||||
assert queued == [src_id]
|
||||
assert fetches == [link_id]
|
||||
|
||||
# The link is armed for a fresh attempt.
|
||||
row = (await db.execute(
|
||||
select(ExternalLink.status, ExternalLink.attempts)
|
||||
.where(ExternalLink.id == link_id)
|
||||
)).one()
|
||||
assert tuple(row) == ("pending", 0)
|
||||
|
||||
# Record gone — the error tombstones cascade away with it.
|
||||
remaining = (await db.execute(
|
||||
|
||||
Reference in New Issue
Block a user