Compare commits
8 Commits
11572f469a
...
f8b667604f
| Author | SHA1 | Date | |
|---|---|---|---|
| f8b667604f | |||
| 4371ddb7e7 | |||
| 40cc11be5b | |||
| 0b78264d62 | |||
| 9eae636047 | |||
| f8105046dc | |||
| d631ed023c | |||
| c64261593d |
@@ -145,6 +145,9 @@ async def similar():
|
|||||||
filters, _sort = _parse_filters()
|
filters, _sort = _parse_filters()
|
||||||
except (KeyError, ValueError):
|
except (KeyError, ValueError):
|
||||||
return jsonify({"error": "similar_to query param required"}), 400
|
return jsonify({"error": "similar_to query param required"}), 400
|
||||||
|
# Explore passes exclude_wip=1 to also drop work-in-progress from the
|
||||||
|
# rabbit-hole; the gallery's own "similar" button omits it (keeps wip, #1274).
|
||||||
|
exclude_wip = request.args.get("exclude_wip") in ("1", "true", "True")
|
||||||
# post_id is the exclusive post-detail view — not a similarity scope.
|
# post_id is the exclusive post-detail view — not a similarity scope.
|
||||||
# include_hidden is a gallery-browse flag; similar() has its OWN presentation
|
# include_hidden is a gallery-browse flag; similar() has its OWN presentation
|
||||||
# exclusion (a similarity-quality concern, #1274), so drop it here (#141).
|
# exclusion (a similarity-quality concern, #1274), so drop it here (#141).
|
||||||
@@ -154,7 +157,8 @@ async def similar():
|
|||||||
async with get_session() as session:
|
async with get_session() as session:
|
||||||
svc = GalleryService(session)
|
svc = GalleryService(session)
|
||||||
try:
|
try:
|
||||||
images = await svc.similar(image_id=similar_to, limit=limit, **scope)
|
images = await svc.similar(
|
||||||
|
image_id=similar_to, limit=limit, exclude_wip=exclude_wip, **scope)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
return jsonify({"error": str(exc)}), 400
|
return jsonify({"error": str(exc)}), 400
|
||||||
if images is None:
|
if images is None:
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ from ..models import (
|
|||||||
ImportTask,
|
ImportTask,
|
||||||
Post,
|
Post,
|
||||||
Tag,
|
Tag,
|
||||||
|
TaskRun,
|
||||||
)
|
)
|
||||||
from ..services import interpreter_client as ic
|
from ..services import interpreter_client as ic
|
||||||
|
|
||||||
@@ -306,6 +307,10 @@ async def translation_status():
|
|||||||
"""For the Settings card: is it on, is a URL set, is the service reachable,
|
"""For the Settings card: is it on, is a URL set, is the service reachable,
|
||||||
and how many posts still await translation. Health runs the sync client in a
|
and how many posts still await translation. Health runs the sync client in a
|
||||||
thread so the event loop isn't blocked."""
|
thread so the event loop isn't blocked."""
|
||||||
|
translation_tasks = (
|
||||||
|
"backend.app.tasks.translation.translate_posts",
|
||||||
|
"backend.app.tasks.translation.retranslate_posts",
|
||||||
|
)
|
||||||
async with get_session() as session:
|
async with get_session() as session:
|
||||||
cfg = await ImportSettings.load(session)
|
cfg = await ImportSettings.load(session)
|
||||||
untranslated = (await session.execute(
|
untranslated = (await session.execute(
|
||||||
@@ -315,6 +320,21 @@ async def translation_status():
|
|||||||
Post.post_title.is_not(None), Post.description.is_not(None),
|
Post.post_title.is_not(None), Post.description.is_not(None),
|
||||||
))
|
))
|
||||||
)).scalar_one()
|
)).scalar_one()
|
||||||
|
# Live progress: is a sweep running now, and what did the last one do?
|
||||||
|
# (run-until-done re-enqueues itself, so `active` stays true across a
|
||||||
|
# bulk re-translate; `last_run` surfaces a completed run's outcome.)
|
||||||
|
active = (await session.execute(
|
||||||
|
select(func.count(TaskRun.id))
|
||||||
|
.where(TaskRun.task_name.in_(translation_tasks))
|
||||||
|
.where(TaskRun.status == "running")
|
||||||
|
)).scalar_one()
|
||||||
|
last = (await session.execute(
|
||||||
|
select(TaskRun.task_name, TaskRun.status, TaskRun.finished_at)
|
||||||
|
.where(TaskRun.task_name.in_(translation_tasks))
|
||||||
|
.where(TaskRun.finished_at.is_not(None))
|
||||||
|
.order_by(TaskRun.finished_at.desc())
|
||||||
|
.limit(1)
|
||||||
|
)).first()
|
||||||
base_url = (cfg.interpreter_base_url or "").strip()
|
base_url = (cfg.interpreter_base_url or "").strip()
|
||||||
healthy = await asyncio.to_thread(ic.health, base_url) if base_url else False
|
healthy = await asyncio.to_thread(ic.health, base_url) if base_url else False
|
||||||
return jsonify({
|
return jsonify({
|
||||||
@@ -322,6 +342,12 @@ async def translation_status():
|
|||||||
"base_url_set": bool(base_url),
|
"base_url_set": bool(base_url),
|
||||||
"healthy": healthy,
|
"healthy": healthy,
|
||||||
"untranslated_count": int(untranslated),
|
"untranslated_count": int(untranslated),
|
||||||
|
"active": int(active) > 0,
|
||||||
|
"last_run": {
|
||||||
|
"task": last[0].rsplit(".", 1)[-1],
|
||||||
|
"status": last[1],
|
||||||
|
"finished_at": last[2].isoformat() if last[2] else None,
|
||||||
|
} if last else None,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -70,6 +70,14 @@ def make_celery() -> Celery:
|
|||||||
},
|
},
|
||||||
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
|
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
|
||||||
task_acks_late=True,
|
task_acks_late=True,
|
||||||
|
# Deploy graceful-shutdown safety: with acks_late, a task killed because
|
||||||
|
# it outran the container's stop-grace window (SIGKILL) is re-queued
|
||||||
|
# rather than silently lost. Safe because our long tasks are idempotent +
|
||||||
|
# chunked (translation per-post commit, downloads terminal-status, audits
|
||||||
|
# chunk) and the 5-min recovery sweeps re-drive anything left non-terminal
|
||||||
|
# — a re-run resumes cleanly and never corrupts. No redeliver-loop risk:
|
||||||
|
# heavy GPU work is tombstoned via gpu_queue, not run inline in a worker.
|
||||||
|
task_reject_on_worker_lost=True,
|
||||||
worker_prefetch_multiplier=1,
|
worker_prefetch_multiplier=1,
|
||||||
# Broker resilience (2026-06-24): a swarm overlay-network blip after a
|
# Broker resilience (2026-06-24): a swarm overlay-network blip after a
|
||||||
# redeploy left Redis healthy but transiently unreachable, and a worker
|
# redeploy left Redis healthy but transiently unreachable, and a worker
|
||||||
@@ -107,6 +115,11 @@ def make_celery() -> Celery:
|
|||||||
"task": "backend.app.tasks.maintenance.cleanup_old_tasks",
|
"task": "backend.app.tasks.maintenance.cleanup_old_tasks",
|
||||||
"schedule": 86400.0, # daily
|
"schedule": 86400.0, # daily
|
||||||
},
|
},
|
||||||
|
"cleanup-orphaned-temp-files": {
|
||||||
|
"task": "backend.app.tasks.maintenance.cleanup_orphaned_temp_files",
|
||||||
|
"schedule": 86400.0, # daily — sweep .part/.partial left by a
|
||||||
|
# download/import killed mid-write (graceful-shutdown fallout)
|
||||||
|
},
|
||||||
"train-heads-nightly": {
|
"train-heads-nightly": {
|
||||||
"task": "backend.app.tasks.ml.scheduled_train_heads",
|
"task": "backend.app.tasks.ml.scheduled_train_heads",
|
||||||
"schedule": 86400.0, # passive cadence; manual retrain stays available
|
"schedule": 86400.0, # passive cadence; manual retrain stays available
|
||||||
|
|||||||
@@ -48,6 +48,10 @@ class TagKind(StrEnum):
|
|||||||
# content. `wip` is real art: only the training pipelines exclude it.
|
# content. `wip` is real art: only the training pipelines exclude it.
|
||||||
SYSTEM_TAG_NAMES = ("wip", "banner", "editor screenshot")
|
SYSTEM_TAG_NAMES = ("wip", "banner", "editor screenshot")
|
||||||
PRESENTATION_SYSTEM_TAGS = ("banner", "editor screenshot")
|
PRESENTATION_SYSTEM_TAGS = ("banner", "editor screenshot")
|
||||||
|
# `wip` marks real-but-unfinished art. It's kept in the gallery's own "similar"
|
||||||
|
# results (#1274), but the Explore rabbit-hole opts to hide it (exclude_wip) so a
|
||||||
|
# browse doesn't keep surfacing work-in-progress (operator, 2026-07-08).
|
||||||
|
WIP_SYSTEM_TAG = "wip"
|
||||||
|
|
||||||
image_tag = Table(
|
image_tag = Table(
|
||||||
"image_tag",
|
"image_tag",
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ from ..models import (
|
|||||||
Tag,
|
Tag,
|
||||||
TagPositiveConfirmation,
|
TagPositiveConfirmation,
|
||||||
)
|
)
|
||||||
from ..models.tag import PRESENTATION_SYSTEM_TAGS, image_tag
|
from ..models.tag import PRESENTATION_SYSTEM_TAGS, WIP_SYSTEM_TAG, image_tag
|
||||||
from .pagination import decode_cursor, encode_cursor
|
from .pagination import decode_cursor, encode_cursor
|
||||||
from .tag_query import (
|
from .tag_query import (
|
||||||
fandom_join_alias,
|
fandom_join_alias,
|
||||||
@@ -715,6 +715,7 @@ class GalleryService:
|
|||||||
platform: str | None = None,
|
platform: str | None = None,
|
||||||
untagged: bool = False, no_artist: bool = False,
|
untagged: bool = False, no_artist: bool = False,
|
||||||
date_from: datetime | None = None, date_to: datetime | None = None,
|
date_from: datetime | None = None, date_to: datetime | None = None,
|
||||||
|
exclude_wip: bool = False,
|
||||||
) -> list[GalleryImage] | None:
|
) -> list[GalleryImage] | None:
|
||||||
"""Visual "more like this": images near `image_id`'s SigLIP embedding
|
"""Visual "more like this": images near `image_id`'s SigLIP embedding
|
||||||
(pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result
|
(pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result
|
||||||
@@ -751,14 +752,18 @@ class GalleryService:
|
|||||||
# Presentation images (banner / editor-screenshot system tags, #128)
|
# Presentation images (banner / editor-screenshot system tags, #128)
|
||||||
# cluster on UI chrome rather than content, so near any one of them
|
# cluster on UI chrome rather than content, so near any one of them
|
||||||
# they'd fill the grid. Excluded from CANDIDATES only — the anchor
|
# they'd fill the grid. Excluded from CANDIDATES only — the anchor
|
||||||
# itself may be a banner, and `wip` stays surfaced (real art; only
|
# itself may be a banner. `wip` stays surfaced here by default (real art;
|
||||||
# the training pipelines exclude it).
|
# only the training pipelines exclude it), but the Explore rabbit-hole
|
||||||
|
# passes exclude_wip to also drop work-in-progress (operator, 2026-07-08).
|
||||||
|
excluded_system_tags = PRESENTATION_SYSTEM_TAGS
|
||||||
|
if exclude_wip:
|
||||||
|
excluded_system_tags = (*PRESENTATION_SYSTEM_TAGS, WIP_SYSTEM_TAG)
|
||||||
presentation = (
|
presentation = (
|
||||||
select(image_tag.c.image_record_id)
|
select(image_tag.c.image_record_id)
|
||||||
.join(Tag, Tag.id == image_tag.c.tag_id)
|
.join(Tag, Tag.id == image_tag.c.tag_id)
|
||||||
.where(
|
.where(
|
||||||
Tag.is_system.is_(True),
|
Tag.is_system.is_(True),
|
||||||
Tag.name.in_(PRESENTATION_SYSTEM_TAGS),
|
Tag.name.in_(excluded_system_tags),
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
stmt = stmt.where(
|
stmt = stmt.where(
|
||||||
|
|||||||
@@ -8,12 +8,24 @@ behind a reverse proxy). Full API contract: Scribe note #1347.
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
from email.utils import parsedate_to_datetime
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
from requests.adapters import HTTPAdapter
|
||||||
|
from urllib3.util.retry import Retry
|
||||||
|
|
||||||
|
|
||||||
class InterpreterUnavailable(Exception):
|
class InterpreterUnavailable(Exception):
|
||||||
"""The translation engine is down / unreachable (HTTP 503 or a connection
|
"""The translation engine is down / unreachable / draining — a connection
|
||||||
error) — retry later, don't drop the item."""
|
error, HTTP 429, or a 5xx (commonly 502/503/504 through a reverse proxy while
|
||||||
|
the service restarts). Retry later, don't drop the item. ``retry_after`` holds
|
||||||
|
the server's Retry-After hint in seconds when it sent one, so the caller can
|
||||||
|
back off exactly as long as it's asked to."""
|
||||||
|
|
||||||
|
def __init__(self, message: str, *, retry_after: float | None = None):
|
||||||
|
super().__init__(message)
|
||||||
|
self.retry_after = retry_after
|
||||||
|
|
||||||
|
|
||||||
class InterpreterBadRequest(Exception):
|
class InterpreterBadRequest(Exception):
|
||||||
@@ -24,13 +36,52 @@ def _url(base_url: str, path: str) -> str:
|
|||||||
return f"{base_url.rstrip('/')}{path}"
|
return f"{base_url.rstrip('/')}{path}"
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_retry_after(resp) -> float | None:
|
||||||
|
"""Parse a Retry-After header (RFC 7231: delta-seconds or an HTTP-date) into
|
||||||
|
non-negative seconds, or None if absent/unparseable — so a gracefully-
|
||||||
|
draining Interpreter can tell Curator exactly how long to wait before it
|
||||||
|
tries again."""
|
||||||
|
raw = (resp.headers.get("Retry-After") or "").strip()
|
||||||
|
if not raw:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return max(0.0, float(int(raw))) # delta-seconds form
|
||||||
|
except ValueError:
|
||||||
|
pass
|
||||||
|
try: # HTTP-date form
|
||||||
|
when = parsedate_to_datetime(raw)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
if when is None:
|
||||||
|
return None
|
||||||
|
if when.tzinfo is None:
|
||||||
|
when = when.replace(tzinfo=UTC)
|
||||||
|
return max(0.0, (when - datetime.now(UTC)).total_seconds())
|
||||||
|
|
||||||
|
|
||||||
|
# A shared session pools the keep-alive connection across the per-post sweep
|
||||||
|
# calls, and retries CONNECT failures only (connect=2, short backoff) — smoothing
|
||||||
|
# the instant a reverse proxy reloads. Status codes are deliberately NOT retried
|
||||||
|
# (status=0, raise_on_status=False): translate() maps 429/5xx → InterpreterUnavailable
|
||||||
|
# itself, and letting urllib3 retry a draining 503 would defeat the Retry-After
|
||||||
|
# backoff we honour upstream.
|
||||||
|
_retry = Retry(
|
||||||
|
total=None, connect=2, read=0, redirect=0, status=0,
|
||||||
|
backoff_factor=0.3, raise_on_status=False,
|
||||||
|
)
|
||||||
|
session = requests.Session()
|
||||||
|
_adapter = HTTPAdapter(max_retries=_retry)
|
||||||
|
session.mount("http://", _adapter)
|
||||||
|
session.mount("https://", _adapter)
|
||||||
|
|
||||||
|
|
||||||
def health(base_url: str, *, timeout: float = 5.0) -> bool:
|
def health(base_url: str, *, timeout: float = 5.0) -> bool:
|
||||||
"""True iff the Interpreter LLM engine is up. Any error (unset URL, network,
|
"""True iff the Interpreter LLM engine is up. Any error (unset URL, network,
|
||||||
non-200, engine down) → False, so the sweep just no-ops rather than raising."""
|
non-200, engine down) → False, so the sweep just no-ops rather than raising."""
|
||||||
if not base_url:
|
if not base_url:
|
||||||
return False
|
return False
|
||||||
try:
|
try:
|
||||||
r = requests.get(_url(base_url, "/v1/health"), timeout=timeout)
|
r = session.get(_url(base_url, "/v1/health"), timeout=timeout)
|
||||||
except requests.RequestException:
|
except requests.RequestException:
|
||||||
return False
|
return False
|
||||||
if r.status_code != 200:
|
if r.status_code != 200:
|
||||||
@@ -53,14 +104,15 @@ def translate(
|
|||||||
Interpreter batch metadata is aggregate (first item only) — fine for one
|
Interpreter batch metadata is aggregate (first item only) — fine for one
|
||||||
post's ``[title, description]`` batch since they share a language. Passthrough
|
post's ``[title, description]`` batch since they share a language. Passthrough
|
||||||
items (already target-language / emoji-only) come back UNCHANGED in their
|
items (already target-language / emoji-only) come back UNCHANGED in their
|
||||||
slot. Raises InterpreterUnavailable on 503 / connection error (retry later),
|
slot. Raises InterpreterUnavailable on a connection error / 429 / 5xx (retry
|
||||||
InterpreterBadRequest on 400. (Scribe note #1347.)
|
later, honouring Retry-After), InterpreterBadRequest on 400. (Scribe note
|
||||||
|
#1347.)
|
||||||
"""
|
"""
|
||||||
if not texts:
|
if not texts:
|
||||||
return {"translations": [], "detected_lang": None,
|
return {"translations": [], "detected_lang": None,
|
||||||
"engine": None, "engine_version": None}
|
"engine": None, "engine_version": None}
|
||||||
try:
|
try:
|
||||||
r = requests.post(
|
r = session.post(
|
||||||
_url(base_url, "/v1/translate"),
|
_url(base_url, "/v1/translate"),
|
||||||
json={"q": list(texts), "source": source,
|
json={"q": list(texts), "source": source,
|
||||||
"target": target, "engine": "auto"},
|
"target": target, "engine": "auto"},
|
||||||
@@ -68,10 +120,18 @@ def translate(
|
|||||||
)
|
)
|
||||||
except requests.RequestException as e:
|
except requests.RequestException as e:
|
||||||
raise InterpreterUnavailable(str(e)) from e
|
raise InterpreterUnavailable(str(e)) from e
|
||||||
if r.status_code == 503:
|
|
||||||
raise InterpreterUnavailable("interpreter engine unavailable")
|
|
||||||
if r.status_code == 400:
|
if r.status_code == 400:
|
||||||
raise InterpreterBadRequest((r.json() or {}).get("error", "bad request"))
|
raise InterpreterBadRequest((r.json() or {}).get("error", "bad request"))
|
||||||
|
# A gracefully-draining service — often behind a reverse proxy — returns 429
|
||||||
|
# or a 5xx gateway error (502/503/504), not always a clean 503. Treat every
|
||||||
|
# one as "unavailable, retry later" and honour any Retry-After it sends, so a
|
||||||
|
# rolling Interpreter restart interrupts the sweep cleanly (instead of raising
|
||||||
|
# an opaque HTTPError) and resumes exactly when the service says it's ready.
|
||||||
|
if r.status_code == 429 or r.status_code >= 500:
|
||||||
|
raise InterpreterUnavailable(
|
||||||
|
f"interpreter unavailable (HTTP {r.status_code})",
|
||||||
|
retry_after=_parse_retry_after(r),
|
||||||
|
)
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
body = r.json() or {}
|
body = r.json() or {}
|
||||||
out = body.get("translatedText")
|
out = body.get("translatedText")
|
||||||
|
|||||||
@@ -79,6 +79,14 @@ VERIFY_PAGE = 200
|
|||||||
FFPROBE_TIMEOUT_SECONDS = 10
|
FFPROBE_TIMEOUT_SECONDS = 10
|
||||||
TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h
|
TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h
|
||||||
TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days
|
TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days
|
||||||
|
# Orphaned staging files: downloads/imports stage into <name>.part / <name>.partial
|
||||||
|
# then os.replace() into place (importer / external_fetch / native_ingest_common /
|
||||||
|
# attachment_store), so a kill mid-write leaves a discardable temp, never a corrupt
|
||||||
|
# final. cleanup_orphaned_temp_files sweeps ones left behind; the min-age guard
|
||||||
|
# keeps it from deleting an in-flight download's staging file mid-write.
|
||||||
|
IMAGES_ROOT = Path("/images")
|
||||||
|
TEMP_STAGING_SUFFIXES = (".part", ".partial")
|
||||||
|
ORPHAN_TEMP_MIN_AGE_HOURS = 6
|
||||||
|
|
||||||
# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be
|
# Audit 2026-06-02: per-entity recovery sweep thresholds. Each must be
|
||||||
# > the entity's longest legitimate runtime (its task's time_limit + a
|
# > the entity's longest legitimate runtime (its task's time_limit + a
|
||||||
@@ -339,6 +347,34 @@ def cleanup_old_tasks() -> int:
|
|||||||
return result.rowcount or 0
|
return result.rowcount or 0
|
||||||
|
|
||||||
|
|
||||||
|
@celery.task(name="backend.app.tasks.maintenance.cleanup_orphaned_temp_files")
|
||||||
|
def cleanup_orphaned_temp_files() -> int:
|
||||||
|
"""Delete orphaned .part/.partial staging files under the images root, left by
|
||||||
|
a download/import killed mid-write. Only removes files older than
|
||||||
|
ORPHAN_TEMP_MIN_AGE_HOURS so an in-flight download's staging file is never
|
||||||
|
pulled out from under it. Returns the count removed."""
|
||||||
|
if not IMAGES_ROOT.is_dir():
|
||||||
|
return 0
|
||||||
|
cutoff = datetime.now(UTC).timestamp() - ORPHAN_TEMP_MIN_AGE_HOURS * 3600
|
||||||
|
removed = 0
|
||||||
|
for path in IMAGES_ROOT.rglob("*"):
|
||||||
|
if path.suffix not in TEMP_STAGING_SUFFIXES or not path.is_file():
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
if path.stat().st_mtime >= cutoff:
|
||||||
|
continue # too fresh — may be an active download
|
||||||
|
path.unlink()
|
||||||
|
removed += 1
|
||||||
|
except OSError as exc:
|
||||||
|
log.warning("cleanup_orphaned_temp_files: %s: %s", path, exc)
|
||||||
|
if removed:
|
||||||
|
log.info(
|
||||||
|
"cleanup_orphaned_temp_files: removed %d orphaned staging file(s)",
|
||||||
|
removed,
|
||||||
|
)
|
||||||
|
return removed
|
||||||
|
|
||||||
|
|
||||||
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
|
@celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
|
||||||
def recover_stalled_task_runs() -> int:
|
def recover_stalled_task_runs() -> int:
|
||||||
"""Flip task_run rows stuck in 'running' past their queue-specific
|
"""Flip task_run rows stuck in 'running' past their queue-specific
|
||||||
|
|||||||
@@ -36,6 +36,20 @@ _MAX_POSTS_PER_RUN = 300
|
|||||||
# Short gap between run-until-done chunks so a big re-translate finishes on its
|
# Short gap between run-until-done chunks so a big re-translate finishes on its
|
||||||
# own without monopolising the maintenance_long worker in one uninterrupted run.
|
# own without monopolising the maintenance_long worker in one uninterrupted run.
|
||||||
_RETRANSLATE_COUNTDOWN = 5
|
_RETRANSLATE_COUNTDOWN = 5
|
||||||
|
# When Interpreter drains mid-chunk (graceful restart), resume the bulk
|
||||||
|
# re-translate after the service's Retry-After hint if it sent one, else this
|
||||||
|
# default; capped so a bogus/huge hint can't park the work longer than the daily
|
||||||
|
# beat's own safety net would.
|
||||||
|
_INTERRUPT_BACKOFF = 60
|
||||||
|
_INTERRUPT_BACKOFF_MAX = 900
|
||||||
|
|
||||||
|
|
||||||
|
def _interrupt_backoff(retry_after) -> int:
|
||||||
|
"""Seconds to wait before resuming after an Interpreter interruption: the
|
||||||
|
server's Retry-After hint (capped), else the default."""
|
||||||
|
if retry_after and retry_after > 0:
|
||||||
|
return int(min(retry_after, _INTERRUPT_BACKOFF_MAX))
|
||||||
|
return _INTERRUPT_BACKOFF
|
||||||
|
|
||||||
|
|
||||||
@celery.task(
|
@celery.task(
|
||||||
@@ -56,7 +70,21 @@ def translate_posts() -> str:
|
|||||||
return ready
|
return ready
|
||||||
base_url, target = ready
|
base_url, target = ready
|
||||||
posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN)
|
posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN)
|
||||||
status, translated = _translate_batch(session, posts, base_url, target)
|
status, translated, retry_after = _translate_batch(
|
||||||
|
session, posts, base_url, target,
|
||||||
|
)
|
||||||
|
# Steady-state daily sweep: one chunk per run on success (the beat drives
|
||||||
|
# the rest — no tail-chase). But if Interpreter drained mid-chunk, don't
|
||||||
|
# make a manual "Translate now" wait a whole day — re-enqueue after its
|
||||||
|
# Retry-After hint (or the default backoff). Self-terminating: once the
|
||||||
|
# service is down the health gate returns "interpreter unavailable" early.
|
||||||
|
if status == "interrupted" and _count_untranslated(session, None):
|
||||||
|
delay = _interrupt_backoff(retry_after)
|
||||||
|
translate_posts.apply_async((), {}, countdown=delay)
|
||||||
|
log.info(
|
||||||
|
"translate_posts: interrupted (service draining) → retry in %ss",
|
||||||
|
delay,
|
||||||
|
)
|
||||||
return _summary(status, translated, len(posts))
|
return _summary(status, translated, len(posts))
|
||||||
|
|
||||||
|
|
||||||
@@ -88,11 +116,11 @@ def retranslate_posts(artist_ids=None, _reset_done=False) -> str:
|
|||||||
)
|
)
|
||||||
|
|
||||||
posts = _select_untranslated(session, artist_ids, _MAX_POSTS_PER_RUN)
|
posts = _select_untranslated(session, artist_ids, _MAX_POSTS_PER_RUN)
|
||||||
status, translated = _translate_batch(session, posts, base_url, target)
|
status, translated, retry_after = _translate_batch(
|
||||||
|
session, posts, base_url, target,
|
||||||
|
)
|
||||||
|
|
||||||
# run-until-done: only chase the tail when the chunk finished cleanly. A
|
# run-until-done: chase the tail when the chunk finished cleanly.
|
||||||
# service interruption / bad request stops here and leaves the rest NULL
|
|
||||||
# for the daily beat or a fresh trigger to resume (rule-89 recovery).
|
|
||||||
# Termination is guaranteed: every handled post leaves the NULL set, so
|
# Termination is guaranteed: every handled post leaves the NULL set, so
|
||||||
# the remaining count strictly decreases each chunk.
|
# the remaining count strictly decreases each chunk.
|
||||||
if status == "ok":
|
if status == "ok":
|
||||||
@@ -106,6 +134,23 @@ def retranslate_posts(artist_ids=None, _reset_done=False) -> str:
|
|||||||
log.info(
|
log.info(
|
||||||
"retranslate_posts: %d remaining → re-enqueued", remaining,
|
"retranslate_posts: %d remaining → re-enqueued", remaining,
|
||||||
)
|
)
|
||||||
|
# Interpreter drained mid-chunk (graceful restart / 429 / 5xx): don't
|
||||||
|
# stall the bulk re-translate until the daily beat — resume it after the
|
||||||
|
# service's Retry-After hint (or the default backoff), with
|
||||||
|
# _reset_done=True so the fresh work is never re-wiped. Self-terminating:
|
||||||
|
# if the service is still down next run, _translation_config's health gate
|
||||||
|
# returns "interpreter unavailable" early and nothing re-enqueues.
|
||||||
|
elif status == "interrupted" and _count_untranslated(session, artist_ids):
|
||||||
|
delay = _interrupt_backoff(retry_after)
|
||||||
|
retranslate_posts.apply_async(
|
||||||
|
(),
|
||||||
|
{"artist_ids": artist_ids, "_reset_done": True},
|
||||||
|
countdown=delay,
|
||||||
|
)
|
||||||
|
log.info(
|
||||||
|
"retranslate_posts: interrupted (service draining) → retry in %ss",
|
||||||
|
delay,
|
||||||
|
)
|
||||||
return _summary(status, translated, len(posts))
|
return _summary(status, translated, len(posts))
|
||||||
|
|
||||||
|
|
||||||
@@ -174,23 +219,25 @@ def _reset_translations(session, artist_ids) -> int:
|
|||||||
|
|
||||||
def _translate_batch(session, posts, base_url: str, target: str):
|
def _translate_batch(session, posts, base_url: str, target: str):
|
||||||
"""Translate each post in the chunk with a per-post commit (an interrupted
|
"""Translate each post in the chunk with a per-post commit (an interrupted
|
||||||
run keeps progress). Returns (status, translated) where status is one of
|
run keeps progress). Returns (status, translated, retry_after) where status is
|
||||||
"ok" / "interrupted" (503/conn) / "stopped" (400) / "timeout"."""
|
one of "ok" / "interrupted" (unavailable/drain) / "stopped" (400) / "timeout";
|
||||||
|
retry_after is the server's Retry-After hint in seconds on an interrupt, else
|
||||||
|
None."""
|
||||||
translated = 0
|
translated = 0
|
||||||
for post in posts:
|
for post in posts:
|
||||||
try:
|
try:
|
||||||
translated += _translate_one(session, post, base_url, target)
|
translated += _translate_one(session, post, base_url, target)
|
||||||
session.commit()
|
session.commit()
|
||||||
except ic.InterpreterUnavailable:
|
except ic.InterpreterUnavailable as e:
|
||||||
session.rollback()
|
session.rollback()
|
||||||
return "interrupted", translated
|
return "interrupted", translated, getattr(e, "retry_after", None)
|
||||||
except ic.InterpreterBadRequest as e:
|
except ic.InterpreterBadRequest as e:
|
||||||
session.rollback()
|
session.rollback()
|
||||||
log.warning("translate bad request: %s", e)
|
log.warning("translate bad request: %s", e)
|
||||||
return "stopped", translated
|
return "stopped", translated, None
|
||||||
except SoftTimeLimitExceeded:
|
except SoftTimeLimitExceeded:
|
||||||
return "timeout", translated
|
return "timeout", translated, None
|
||||||
return "ok", translated
|
return "ok", translated, None
|
||||||
|
|
||||||
|
|
||||||
def _summary(status: str, translated: int, scanned: int) -> str:
|
def _summary(status: str, translated: int, scanned: int) -> str:
|
||||||
|
|||||||
@@ -8,6 +8,35 @@
|
|||||||
# run `docker compose up` from this directory and switches images to
|
# run `docker compose up` from this directory and switches images to
|
||||||
# local builds + DEBUG logging.
|
# local builds + DEBUG logging.
|
||||||
|
|
||||||
|
# Rolling-deploy safety (Swarm / `docker stack deploy`): update one task at a
|
||||||
|
# time, START the new task before stopping the old (zero-downtime via the ingress
|
||||||
|
# mesh), and if the new task doesn't reach a healthy state within `monitor`, roll
|
||||||
|
# back to the previous image automatically. `monitor` is sized above web's
|
||||||
|
# healthcheck start_period so a broken image that never goes healthy is caught.
|
||||||
|
# Plain `docker compose up` ignores `deploy:` (it warns + skips), so the dev
|
||||||
|
# override is unaffected. Referenced by each long-lived service below.
|
||||||
|
x-deploy-policy: &deploy_policy
|
||||||
|
update_config:
|
||||||
|
order: start-first
|
||||||
|
failure_action: rollback
|
||||||
|
monitor: 90s
|
||||||
|
rollback_config:
|
||||||
|
order: start-first
|
||||||
|
restart_policy:
|
||||||
|
condition: any
|
||||||
|
delay: 10s
|
||||||
|
|
||||||
|
# Worker liveness: ping THIS container's celery node over the broker. Lenient
|
||||||
|
# (60s interval, 3 retries, 60s start_period) so a transient broker blip never
|
||||||
|
# false-flags a worker into a rollback. `$$HOSTNAME` → `$HOSTNAME` for the shell;
|
||||||
|
# celery's default node name is celery@<hostname> (the container id).
|
||||||
|
x-celery-healthcheck: &celery_healthcheck
|
||||||
|
test: ["CMD-SHELL", "celery -A backend.app.celery_app:celery inspect ping -d celery@$$HOSTNAME --timeout 10 >/dev/null 2>&1"]
|
||||||
|
interval: 60s
|
||||||
|
timeout: 15s
|
||||||
|
retries: 3
|
||||||
|
start_period: 60s
|
||||||
|
|
||||||
services:
|
services:
|
||||||
redis:
|
redis:
|
||||||
image: redis:7-alpine
|
image: redis:7-alpine
|
||||||
@@ -47,6 +76,24 @@ services:
|
|||||||
web:
|
web:
|
||||||
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
||||||
command: ["web"]
|
command: ["web"]
|
||||||
|
# Graceful shutdown: give the container time to drain in-flight work on a
|
||||||
|
# deploy (docker SIGTERMs, then SIGKILLs after this window — default is only
|
||||||
|
# 10s, far too short for real jobs). Hypercorn/Celery both warm-shut-down on
|
||||||
|
# SIGTERM; per-lane values sized to typical task length. Anything that still
|
||||||
|
# outruns the window is re-queued (task_reject_on_worker_lost) and re-driven
|
||||||
|
# by the 5-min recovery sweeps, so a kill never corrupts. web = short HTTP
|
||||||
|
# requests + the occasional file download.
|
||||||
|
stop_grace_period: 30s
|
||||||
|
# Liveness for rolling deploys: /api/health is a no-DB 200 (just proves the
|
||||||
|
# app booted + serves HTTP after `alembic upgrade head`). start_period covers
|
||||||
|
# the migration + boot so a slow start isn't mis-flagged.
|
||||||
|
healthcheck:
|
||||||
|
test: ["CMD-SHELL", "python -c \"import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://localhost:8080/api/health', timeout=5).status==200 else 1)\""]
|
||||||
|
interval: 15s
|
||||||
|
timeout: 6s
|
||||||
|
retries: 3
|
||||||
|
start_period: 40s
|
||||||
|
deploy: *deploy_policy
|
||||||
ports:
|
ports:
|
||||||
- "${PORT:-8080}:8080"
|
- "${PORT:-8080}:8080"
|
||||||
environment: &app_env
|
environment: &app_env
|
||||||
@@ -77,6 +124,10 @@ services:
|
|||||||
worker:
|
worker:
|
||||||
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
||||||
command: ["worker"]
|
command: ["worker"]
|
||||||
|
# Drain in-flight import/thumbnail/download tasks before SIGKILL on deploy.
|
||||||
|
stop_grace_period: 90s
|
||||||
|
healthcheck: *celery_healthcheck
|
||||||
|
deploy: *deploy_policy
|
||||||
environment:
|
environment:
|
||||||
<<: *app_env
|
<<: *app_env
|
||||||
CELERY_QUEUES: default,import,thumbnail,download
|
CELERY_QUEUES: default,import,thumbnail,download
|
||||||
@@ -93,6 +144,10 @@ services:
|
|||||||
scheduler:
|
scheduler:
|
||||||
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
||||||
command: ["scheduler"]
|
command: ["scheduler"]
|
||||||
|
# Quick maintenance/scan lane + beat — short tasks, modest drain window.
|
||||||
|
stop_grace_period: 60s
|
||||||
|
healthcheck: *celery_healthcheck
|
||||||
|
deploy: *deploy_policy
|
||||||
environment:
|
environment:
|
||||||
<<: *app_env
|
<<: *app_env
|
||||||
CELERY_QUEUES: maintenance,scan
|
CELERY_QUEUES: maintenance,scan
|
||||||
@@ -110,6 +165,12 @@ services:
|
|||||||
maintenance-long:
|
maintenance-long:
|
||||||
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
image: git.fabledsword.com/bvandeusen/fabledcurator:dev
|
||||||
command: ["worker"]
|
command: ["worker"]
|
||||||
|
# Longest lane (DB backups, library audits, translation backfill) — give it
|
||||||
|
# the most room to finish a chunk gracefully. Chunked + idempotent, so a job
|
||||||
|
# that still outruns this resumes cleanly next run rather than corrupting.
|
||||||
|
stop_grace_period: 180s
|
||||||
|
healthcheck: *celery_healthcheck
|
||||||
|
deploy: *deploy_policy
|
||||||
environment:
|
environment:
|
||||||
<<: *app_env
|
<<: *app_env
|
||||||
CELERY_QUEUES: maintenance_long
|
CELERY_QUEUES: maintenance_long
|
||||||
@@ -125,6 +186,10 @@ services:
|
|||||||
ml-worker:
|
ml-worker:
|
||||||
image: git.fabledsword.com/bvandeusen/fabledcurator-ml:dev
|
image: git.fabledsword.com/bvandeusen/fabledcurator-ml:dev
|
||||||
command: ["ml-worker"]
|
command: ["ml-worker"]
|
||||||
|
# A single GPU inference pass can run tens of seconds — let it finish.
|
||||||
|
stop_grace_period: 120s
|
||||||
|
healthcheck: *celery_healthcheck
|
||||||
|
deploy: *deploy_policy
|
||||||
environment:
|
environment:
|
||||||
<<: *app_env
|
<<: *app_env
|
||||||
volumes:
|
volumes:
|
||||||
|
|||||||
@@ -52,11 +52,25 @@
|
|||||||
prepend-icon="mdi-refresh" :loading="retranslating"
|
prepend-icon="mdi-refresh" :loading="retranslating"
|
||||||
:disabled="!enabled || !baseUrl" @click="confirmAll = true"
|
:disabled="!enabled || !baseUrl" @click="confirmAll = true"
|
||||||
>Re-translate all</v-btn>
|
>Re-translate all</v-btn>
|
||||||
<span v-if="status" class="fc-muted text-caption">
|
<span
|
||||||
|
v-if="status && status.active"
|
||||||
|
class="fc-muted text-caption d-inline-flex align-center" style="gap: 6px;"
|
||||||
|
>
|
||||||
|
<v-progress-circular indeterminate size="12" width="2" color="accent" />
|
||||||
|
Translating… {{ status.untranslated_count }} remaining
|
||||||
|
</span>
|
||||||
|
<span v-else-if="status" class="fc-muted text-caption">
|
||||||
{{ status.untranslated_count }}
|
{{ status.untranslated_count }}
|
||||||
post{{ status.untranslated_count === 1 ? '' : 's' }} awaiting translation
|
post{{ status.untranslated_count === 1 ? '' : 's' }} awaiting translation
|
||||||
</span>
|
</span>
|
||||||
</div>
|
</div>
|
||||||
|
<p
|
||||||
|
v-if="status && status.last_run && ['error', 'timeout'].includes(status.last_run.status)"
|
||||||
|
class="text-caption text-error mt-1 mb-0"
|
||||||
|
>
|
||||||
|
Last translation run ended with “{{ status.last_run.status }}”. It resumes on
|
||||||
|
the next sweep; check the Interpreter connection if it persists.
|
||||||
|
</p>
|
||||||
<p class="fc-muted text-caption mt-2 mb-0">
|
<p class="fc-muted text-caption mt-2 mb-0">
|
||||||
Use <strong>Re-translate all</strong> after switching the Interpreter model —
|
Use <strong>Re-translate all</strong> after switching the Interpreter model —
|
||||||
it clears every stored translation and re-runs it through the new model
|
it clears every stored translation and re-runs it through the new model
|
||||||
@@ -94,7 +108,7 @@
|
|||||||
</template>
|
</template>
|
||||||
|
|
||||||
<script setup>
|
<script setup>
|
||||||
import { computed, onMounted, ref } from 'vue'
|
import { computed, onMounted, onUnmounted, ref } from 'vue'
|
||||||
import { toast } from '../../utils/toast.js'
|
import { toast } from '../../utils/toast.js'
|
||||||
import { useApi } from '../../composables/useApi.js'
|
import { useApi } from '../../composables/useApi.js'
|
||||||
import { useImportStore } from '../../stores/import.js'
|
import { useImportStore } from '../../stores/import.js'
|
||||||
@@ -126,10 +140,16 @@ const statusText = computed(() => {
|
|||||||
return status.value.healthy ? 'Interpreter reachable' : 'Interpreter unreachable'
|
return status.value.healthy ? 'Interpreter reachable' : 'Interpreter unreachable'
|
||||||
})
|
})
|
||||||
|
|
||||||
|
let pollTimer = null
|
||||||
async function loadStatus() {
|
async function loadStatus() {
|
||||||
try { status.value = await api.get('/api/settings/translation/status') }
|
try { status.value = await api.get('/api/settings/translation/status') }
|
||||||
catch { status.value = null }
|
catch { status.value = null }
|
||||||
|
// Poll live while a sweep is running so the remaining count ticks down; stop
|
||||||
|
// when idle (a fresh trigger restarts it via its own setTimeout(loadStatus)).
|
||||||
|
clearTimeout(pollTimer)
|
||||||
|
if (status.value?.active) pollTimer = setTimeout(loadStatus, 3000)
|
||||||
}
|
}
|
||||||
|
onUnmounted(() => clearTimeout(pollTimer))
|
||||||
|
|
||||||
async function onTest() {
|
async function onTest() {
|
||||||
// Ping /v1/health for the CURRENTLY-typed URL (not the saved one), so a new
|
// Ping /v1/health for the CURRENTLY-typed URL (not the saved one), so a new
|
||||||
|
|||||||
@@ -44,7 +44,9 @@ export const useExploreStore = defineStore('explore', () => {
|
|||||||
// empty and let the view explain why (anchor.has_embedding === false).
|
// empty and let the view explain why (anchor.has_embedding === false).
|
||||||
if (detail.has_embedding) {
|
if (detail.has_embedding) {
|
||||||
const body = await api.get('/api/gallery/similar', {
|
const body = await api.get('/api/gallery/similar', {
|
||||||
params: { similar_to: numId, limit: NEIGHBOR_LIMIT },
|
// exclude_wip: keep work-in-progress out of the Explore rabbit-hole
|
||||||
|
// (the gallery's own "similar" button still shows it) — operator 2026-07-08.
|
||||||
|
params: { similar_to: numId, limit: NEIGHBOR_LIMIT, exclude_wip: 1 },
|
||||||
})
|
})
|
||||||
if (!t.isCurrent()) return
|
if (!t.isCurrent()) return
|
||||||
neighbors.value = body.images || []
|
neighbors.value = body.images || []
|
||||||
|
|||||||
@@ -61,6 +61,34 @@ async def test_translation_status_defaults(client):
|
|||||||
assert body["base_url_set"] is False
|
assert body["base_url_set"] is False
|
||||||
assert body["healthy"] is False
|
assert body["healthy"] is False
|
||||||
assert "untranslated_count" in body
|
assert "untranslated_count" in body
|
||||||
|
assert body["active"] is False # no sweep running
|
||||||
|
assert body["last_run"] is None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_translation_status_reports_active_and_last_run(client, db):
|
||||||
|
# A running translate/retranslate TaskRun → active True; the most recent
|
||||||
|
# finished one → last_run (task basename + status).
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
|
||||||
|
from backend.app.models import TaskRun
|
||||||
|
|
||||||
|
db.add(TaskRun(
|
||||||
|
celery_task_id="r-run", queue="maintenance_long",
|
||||||
|
task_name="backend.app.tasks.translation.retranslate_posts",
|
||||||
|
started_at=datetime.now(UTC), status="running",
|
||||||
|
))
|
||||||
|
db.add(TaskRun(
|
||||||
|
celery_task_id="r-done", queue="maintenance_long",
|
||||||
|
task_name="backend.app.tasks.translation.translate_posts",
|
||||||
|
started_at=datetime.now(UTC), finished_at=datetime.now(UTC),
|
||||||
|
status="success",
|
||||||
|
))
|
||||||
|
await db.commit()
|
||||||
|
body = await (await client.get("/api/settings/translation/status")).get_json()
|
||||||
|
assert body["active"] is True
|
||||||
|
assert body["last_run"]["task"] == "translate_posts"
|
||||||
|
assert body["last_run"]["status"] == "success"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|||||||
@@ -98,6 +98,30 @@ async def test_similar_excludes_presentation_tagged_images(db):
|
|||||||
assert {i.id for i in res_from_banner} == {src.id, wipped.id, plain.id}
|
assert {i.id for i in res_from_banner} == {src.id, wipped.id, plain.id}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_similar_exclude_wip_drops_wip_neighbors(db):
|
||||||
|
"""Explore passes exclude_wip=True to also hide work-in-progress from the
|
||||||
|
rabbit-hole (banner is always hidden; wip only when asked)."""
|
||||||
|
src = await _img(db, 1, _vec(1, 0))
|
||||||
|
bannered = await _img(db, 2, _vec(1, 0.02)) # always hidden
|
||||||
|
wipped = await _img(db, 3, _vec(1, 0.3)) # hidden only with exclude_wip
|
||||||
|
plain = await _img(db, 4, _vec(1, 0.6))
|
||||||
|
banner_tag = (await db.execute(select(Tag).where(
|
||||||
|
Tag.is_system.is_(True), Tag.name == "banner"))).scalar_one()
|
||||||
|
wip_tag = (await db.execute(select(Tag).where(
|
||||||
|
Tag.is_system.is_(True), Tag.name == "wip"))).scalar_one()
|
||||||
|
await db.execute(image_tag.insert().values(
|
||||||
|
image_record_id=bannered.id, tag_id=banner_tag.id, source="manual"))
|
||||||
|
await db.execute(image_tag.insert().values(
|
||||||
|
image_record_id=wipped.id, tag_id=wip_tag.id, source="manual"))
|
||||||
|
svc = GalleryService(db)
|
||||||
|
res = await svc.similar(src.id, limit=10, exclude_wip=True)
|
||||||
|
assert [i.id for i in res] == [plain.id] # wip + banner both gone
|
||||||
|
# Default (gallery "similar" button) still keeps wip (#1274).
|
||||||
|
res_default = await svc.similar(src.id, limit=10)
|
||||||
|
assert wipped.id in {i.id for i in res_default}
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_similar_composes_with_tag_filter(db):
|
async def test_similar_composes_with_tag_filter(db):
|
||||||
src = await _img(db, 1, _vec(1, 0))
|
src = await _img(db, 1, _vec(1, 0))
|
||||||
|
|||||||
@@ -6,9 +6,10 @@ from backend.app.services import interpreter_client as ic
|
|||||||
|
|
||||||
|
|
||||||
class _Resp:
|
class _Resp:
|
||||||
def __init__(self, status_code=200, payload=None):
|
def __init__(self, status_code=200, payload=None, headers=None):
|
||||||
self.status_code = status_code
|
self.status_code = status_code
|
||||||
self._payload = payload or {}
|
self._payload = payload or {}
|
||||||
|
self.headers = headers or {}
|
||||||
|
|
||||||
def json(self):
|
def json(self):
|
||||||
return self._payload
|
return self._payload
|
||||||
@@ -29,7 +30,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
|
|||||||
"interpreter": {"engine": "llm", "engine_version": "ollama:x:12b"},
|
"interpreter": {"engine": "llm", "engine_version": "ollama:x:12b"},
|
||||||
})
|
})
|
||||||
|
|
||||||
monkeypatch.setattr(ic.requests, "post", fake_post)
|
monkeypatch.setattr(ic.session, "post", fake_post)
|
||||||
out = ic.translate(["ねこが可愛い", "金髪ギャル!"], base_url="http://i.lan")
|
out = ic.translate(["ねこが可愛い", "金髪ギャル!"], base_url="http://i.lan")
|
||||||
assert out["translations"] == ["The cat is cute", "Blonde gal!"]
|
assert out["translations"] == ["The cat is cute", "Blonde gal!"]
|
||||||
assert out["detected_lang"] == "ja"
|
assert out["detected_lang"] == "ja"
|
||||||
@@ -40,7 +41,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
|
|||||||
|
|
||||||
def test_translate_passthrough_unchanged(monkeypatch):
|
def test_translate_passthrough_unchanged(monkeypatch):
|
||||||
# Already-English items come back unchanged in their slot (engine "none").
|
# Already-English items come back unchanged in their slot (engine "none").
|
||||||
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(200, {
|
monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(200, {
|
||||||
"translatedText": ["already english"],
|
"translatedText": ["already english"],
|
||||||
"detectedLanguage": {"language": "en"},
|
"detectedLanguage": {"language": "en"},
|
||||||
"interpreter": {"engine": "none", "engine_version": None},
|
"interpreter": {"engine": "none", "engine_version": None},
|
||||||
@@ -52,7 +53,7 @@ def test_translate_passthrough_unchanged(monkeypatch):
|
|||||||
|
|
||||||
|
|
||||||
def test_translate_503_raises_unavailable(monkeypatch):
|
def test_translate_503_raises_unavailable(monkeypatch):
|
||||||
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(503, {}))
|
monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(503, {}))
|
||||||
with pytest.raises(ic.InterpreterUnavailable):
|
with pytest.raises(ic.InterpreterUnavailable):
|
||||||
ic.translate(["x"], base_url="http://i.lan")
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
|
|
||||||
@@ -61,14 +62,46 @@ def test_translate_connection_error_raises_unavailable(monkeypatch):
|
|||||||
def boom(*a, **k):
|
def boom(*a, **k):
|
||||||
raise ic.requests.ConnectionError("refused")
|
raise ic.requests.ConnectionError("refused")
|
||||||
|
|
||||||
monkeypatch.setattr(ic.requests, "post", boom)
|
monkeypatch.setattr(ic.session, "post", boom)
|
||||||
with pytest.raises(ic.InterpreterUnavailable):
|
with pytest.raises(ic.InterpreterUnavailable):
|
||||||
ic.translate(["x"], base_url="http://i.lan")
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("code", [429, 500, 502, 503, 504])
|
||||||
|
def test_translate_429_and_5xx_raise_unavailable(monkeypatch, code):
|
||||||
|
# A gracefully-draining service behind a reverse proxy returns 429/502/503/504
|
||||||
|
# — all mean "retry later", not an opaque error.
|
||||||
|
monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(code, {}))
|
||||||
|
with pytest.raises(ic.InterpreterUnavailable) as ei:
|
||||||
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
|
assert ei.value.retry_after is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_translate_honours_retry_after_seconds(monkeypatch):
|
||||||
|
monkeypatch.setattr(
|
||||||
|
ic.session, "post",
|
||||||
|
lambda *a, **k: _Resp(503, {}, headers={"Retry-After": "42"}),
|
||||||
|
)
|
||||||
|
with pytest.raises(ic.InterpreterUnavailable) as ei:
|
||||||
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
|
assert ei.value.retry_after == 42
|
||||||
|
|
||||||
|
|
||||||
|
def test_translate_retry_after_past_http_date_clamps_to_zero(monkeypatch):
|
||||||
|
# HTTP-date form; a past date → non-negative clamp to 0 (deterministic).
|
||||||
|
monkeypatch.setattr(
|
||||||
|
ic.session, "post",
|
||||||
|
lambda *a, **k: _Resp(
|
||||||
|
503, {}, headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}),
|
||||||
|
)
|
||||||
|
with pytest.raises(ic.InterpreterUnavailable) as ei:
|
||||||
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
|
assert ei.value.retry_after == 0.0
|
||||||
|
|
||||||
|
|
||||||
def test_translate_400_raises_bad_request(monkeypatch):
|
def test_translate_400_raises_bad_request(monkeypatch):
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
ic.requests, "post", lambda *a, **k: _Resp(400, {"error": "bad target"})
|
ic.session, "post", lambda *a, **k: _Resp(400, {"error": "bad target"})
|
||||||
)
|
)
|
||||||
with pytest.raises(ic.InterpreterBadRequest):
|
with pytest.raises(ic.InterpreterBadRequest):
|
||||||
ic.translate(["x"], base_url="http://i.lan")
|
ic.translate(["x"], base_url="http://i.lan")
|
||||||
@@ -78,20 +111,20 @@ def test_translate_empty_is_noop(monkeypatch):
|
|||||||
def boom(*a, **k):
|
def boom(*a, **k):
|
||||||
raise AssertionError("should not call the service for an empty batch")
|
raise AssertionError("should not call the service for an empty batch")
|
||||||
|
|
||||||
monkeypatch.setattr(ic.requests, "post", boom)
|
monkeypatch.setattr(ic.session, "post", boom)
|
||||||
assert ic.translate([], base_url="http://i.lan")["translations"] == []
|
assert ic.translate([], base_url="http://i.lan")["translations"] == []
|
||||||
|
|
||||||
|
|
||||||
def test_health_true_when_llm_up(monkeypatch):
|
def test_health_true_when_llm_up(monkeypatch):
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": True}})
|
ic.session, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": True}})
|
||||||
)
|
)
|
||||||
assert ic.health("http://i.lan") is True
|
assert ic.health("http://i.lan") is True
|
||||||
|
|
||||||
|
|
||||||
def test_health_false_when_engine_down(monkeypatch):
|
def test_health_false_when_engine_down(monkeypatch):
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": False}})
|
ic.session, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": False}})
|
||||||
)
|
)
|
||||||
assert ic.health("http://i.lan") is False
|
assert ic.health("http://i.lan") is False
|
||||||
|
|
||||||
@@ -100,7 +133,7 @@ def test_health_false_on_error(monkeypatch):
|
|||||||
def boom(*a, **k):
|
def boom(*a, **k):
|
||||||
raise ic.requests.ConnectionError("refused")
|
raise ic.requests.ConnectionError("refused")
|
||||||
|
|
||||||
monkeypatch.setattr(ic.requests, "get", boom)
|
monkeypatch.setattr(ic.session, "get", boom)
|
||||||
assert ic.health("http://i.lan") is False
|
assert ic.health("http://i.lan") is False
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -26,6 +26,28 @@ def _make_batch(session) -> int:
|
|||||||
return batch.id
|
return batch.id
|
||||||
|
|
||||||
|
|
||||||
|
def test_cleanup_orphaned_temp_files_removes_stale_only(tmp_path, monkeypatch):
|
||||||
|
import os
|
||||||
|
|
||||||
|
from backend.app.tasks import maintenance as m
|
||||||
|
|
||||||
|
monkeypatch.setattr(m, "IMAGES_ROOT", tmp_path)
|
||||||
|
stale = tmp_path / "artist" / "img.jpg.part" # killed download → orphan
|
||||||
|
stale.parent.mkdir(parents=True)
|
||||||
|
stale.write_bytes(b"x")
|
||||||
|
old = datetime.now(UTC).timestamp() - 8 * 3600 # older than the 6h guard
|
||||||
|
os.utime(stale, (old, old))
|
||||||
|
fresh = tmp_path / "in_progress.jpg.partial" # active download → keep
|
||||||
|
fresh.write_bytes(b"x")
|
||||||
|
keep = tmp_path / "real.jpg" # a real image → keep
|
||||||
|
keep.write_bytes(b"x")
|
||||||
|
|
||||||
|
assert m.cleanup_orphaned_temp_files() == 1
|
||||||
|
assert not stale.exists()
|
||||||
|
assert fresh.exists()
|
||||||
|
assert keep.exists()
|
||||||
|
|
||||||
|
|
||||||
def test_recover_interrupted_only_old(db_sync, monkeypatch):
|
def test_recover_interrupted_only_old(db_sync, monkeypatch):
|
||||||
batch_id = _make_batch(db_sync)
|
batch_id = _make_batch(db_sync)
|
||||||
now = datetime.now(UTC)
|
now = datetime.now(UTC)
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import pytest
|
|||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
|
|
||||||
from backend.app.models import Artist, ImportSettings, Post
|
from backend.app.models import Artist, ImportSettings, Post
|
||||||
|
from backend.app.services import interpreter_client as ic
|
||||||
from backend.app.tasks.translation import retranslate_posts, translate_posts
|
from backend.app.tasks.translation import retranslate_posts, translate_posts
|
||||||
|
|
||||||
pytestmark = pytest.mark.integration
|
pytestmark = pytest.mark.integration
|
||||||
@@ -136,6 +137,31 @@ def test_translate_posts_service_down_leaves_untranslated(db_sync, monkeypatch):
|
|||||||
assert p.translated_source_lang is None # will retry next run
|
assert p.translated_source_lang is None # will retry next run
|
||||||
|
|
||||||
|
|
||||||
|
def test_translate_posts_interrupt_reenqueues_after_backoff(db_sync, monkeypatch):
|
||||||
|
# A drain mid-sweep (health passed, translate 503s w/ Retry-After) re-enqueues
|
||||||
|
# the daily sweep after the backoff instead of waiting for tomorrow's beat.
|
||||||
|
_patch(monkeypatch, db_sync)
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
||||||
|
|
||||||
|
def _drain(texts, **k):
|
||||||
|
raise ic.InterpreterUnavailable("draining", retry_after=30)
|
||||||
|
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
|
||||||
|
calls = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.app.tasks.translation.translate_posts.apply_async",
|
||||||
|
lambda *a, **k: calls.append((a, k)),
|
||||||
|
)
|
||||||
|
a = _artist(db_sync)
|
||||||
|
_post(db_sync, a.id, title="ねこ", ext="p1")
|
||||||
|
_enable(db_sync)
|
||||||
|
db_sync.commit()
|
||||||
|
|
||||||
|
assert "interrupted" in translate_posts()
|
||||||
|
assert len(calls) == 1
|
||||||
|
assert calls[0][1]["countdown"] == 30
|
||||||
|
|
||||||
|
|
||||||
def _mock_new_model(monkeypatch, *, ver="v2"):
|
def _mock_new_model(monkeypatch, *, ver="v2"):
|
||||||
"""Interpreter is up and translates with a NEW engine version."""
|
"""Interpreter is up and translates with a NEW engine version."""
|
||||||
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
||||||
@@ -239,3 +265,61 @@ def test_retranslate_unhealthy_does_not_reset(db_sync, monkeypatch):
|
|||||||
db_sync.refresh(p)
|
db_sync.refresh(p)
|
||||||
assert p.translated_source_lang == "ja"
|
assert p.translated_source_lang == "ja"
|
||||||
assert p.post_title_translated == "OLD"
|
assert p.post_title_translated == "OLD"
|
||||||
|
|
||||||
|
|
||||||
|
def test_retranslate_interrupt_resumes_after_retry_after(db_sync, monkeypatch):
|
||||||
|
# Interpreter drains mid-chunk (health passed, then translate 503s with a
|
||||||
|
# Retry-After): the bulk re-translate re-enqueues itself after the hinted
|
||||||
|
# backoff, with _reset_done=True so it never re-wipes fresh work — instead of
|
||||||
|
# stalling until the daily beat.
|
||||||
|
_patch(monkeypatch, db_sync)
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
||||||
|
|
||||||
|
def _drain(texts, **k):
|
||||||
|
raise ic.InterpreterUnavailable("draining", retry_after=42)
|
||||||
|
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
|
||||||
|
calls = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.app.tasks.translation.retranslate_posts.apply_async",
|
||||||
|
lambda *a, **k: calls.append((a, k)),
|
||||||
|
)
|
||||||
|
a = _artist(db_sync)
|
||||||
|
p = _post(db_sync, a.id, title="ねこ", ext="p1")
|
||||||
|
_mark_translated(p)
|
||||||
|
_enable(db_sync)
|
||||||
|
db_sync.commit()
|
||||||
|
|
||||||
|
assert "interrupted" in retranslate_posts(artist_ids=[a.id])
|
||||||
|
assert len(calls) == 1 # resume re-enqueued
|
||||||
|
args, kw = calls[0]
|
||||||
|
assert args[1]["_reset_done"] is True
|
||||||
|
assert args[1]["artist_ids"] == [a.id]
|
||||||
|
assert kw["countdown"] == 42 # honoured the Retry-After
|
||||||
|
db_sync.refresh(p)
|
||||||
|
assert p.translated_source_lang is None # reset happened, not re-run
|
||||||
|
|
||||||
|
|
||||||
|
def test_retranslate_interrupt_default_backoff_without_hint(db_sync, monkeypatch):
|
||||||
|
# No Retry-After header → fall back to the default backoff (60s).
|
||||||
|
_patch(monkeypatch, db_sync)
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
||||||
|
|
||||||
|
def _drain(texts, **k):
|
||||||
|
raise ic.InterpreterUnavailable("draining")
|
||||||
|
|
||||||
|
monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
|
||||||
|
calls = []
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.app.tasks.translation.retranslate_posts.apply_async",
|
||||||
|
lambda *a, **k: calls.append((a, k)),
|
||||||
|
)
|
||||||
|
a = _artist(db_sync)
|
||||||
|
p = _post(db_sync, a.id, title="ねこ", ext="p1")
|
||||||
|
_mark_translated(p)
|
||||||
|
_enable(db_sync)
|
||||||
|
db_sync.commit()
|
||||||
|
|
||||||
|
retranslate_posts(artist_ids=[a.id])
|
||||||
|
assert len(calls) == 1
|
||||||
|
assert calls[0][1]["countdown"] == 60
|
||||||
|
|||||||
Reference in New Issue
Block a user