diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 0e47473..fb15685 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -70,6 +70,14 @@ def make_celery() -> Celery: }, # Heavy ML tasks need fair dispatch — see ImageRepo's precedent. 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, # Broker resilience (2026-06-24): a swarm overlay-network blip after a # redeploy left Redis healthy but transiently unreachable, and a worker diff --git a/backend/app/services/interpreter_client.py b/backend/app/services/interpreter_client.py index 0e02029..70008d2 100644 --- a/backend/app/services/interpreter_client.py +++ b/backend/app/services/interpreter_client.py @@ -8,12 +8,22 @@ behind a reverse proxy). Full API contract: Scribe note #1347. """ from __future__ import annotations +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime + import requests class InterpreterUnavailable(Exception): - """The translation engine is down / unreachable (HTTP 503 or a connection - error) — retry later, don't drop the item.""" + """The translation engine is down / unreachable / draining — a connection + 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): @@ -24,6 +34,29 @@ def _url(base_url: str, path: str) -> str: 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=timezone.utc) + return max(0.0, (when - datetime.now(timezone.utc)).total_seconds()) + + def health(base_url: str, *, timeout: float = 5.0) -> bool: """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.""" @@ -53,8 +86,9 @@ def translate( Interpreter batch metadata is aggregate (first item only) — fine for one post's ``[title, description]`` batch since they share a language. Passthrough items (already target-language / emoji-only) come back UNCHANGED in their - slot. Raises InterpreterUnavailable on 503 / connection error (retry later), - InterpreterBadRequest on 400. (Scribe note #1347.) + slot. Raises InterpreterUnavailable on a connection error / 429 / 5xx (retry + later, honouring Retry-After), InterpreterBadRequest on 400. (Scribe note + #1347.) """ if not texts: return {"translations": [], "detected_lang": None, @@ -68,10 +102,18 @@ def translate( ) except requests.RequestException as e: raise InterpreterUnavailable(str(e)) from e - if r.status_code == 503: - raise InterpreterUnavailable("interpreter engine unavailable") if r.status_code == 400: 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() body = r.json() or {} out = body.get("translatedText") diff --git a/backend/app/tasks/translation.py b/backend/app/tasks/translation.py index 15dd499..70f5b51 100644 --- a/backend/app/tasks/translation.py +++ b/backend/app/tasks/translation.py @@ -36,6 +36,20 @@ _MAX_POSTS_PER_RUN = 300 # 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. _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( @@ -56,7 +70,10 @@ def translate_posts() -> str: return ready base_url, target = ready posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN) - status, translated = _translate_batch(session, posts, base_url, target) + # Steady-state daily sweep: one chunk per run. On interruption the + # untranslated rows stay NULL and the next daily beat (or a fresh + # "Translate now") resumes them — no self-chase needed here. + status, translated, _retry = _translate_batch(session, posts, base_url, target) return _summary(status, translated, len(posts)) @@ -88,11 +105,11 @@ def retranslate_posts(artist_ids=None, _reset_done=False) -> str: ) 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 - # service interruption / bad request stops here and leaves the rest NULL - # for the daily beat or a fresh trigger to resume (rule-89 recovery). + # run-until-done: chase the tail when the chunk finished cleanly. # Termination is guaranteed: every handled post leaves the NULL set, so # the remaining count strictly decreases each chunk. if status == "ok": @@ -106,6 +123,23 @@ def retranslate_posts(artist_ids=None, _reset_done=False) -> str: log.info( "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)) @@ -174,23 +208,25 @@ def _reset_translations(session, artist_ids) -> int: def _translate_batch(session, posts, base_url: str, target: str): """Translate each post in the chunk with a per-post commit (an interrupted - run keeps progress). Returns (status, translated) where status is one of - "ok" / "interrupted" (503/conn) / "stopped" (400) / "timeout".""" + run keeps progress). Returns (status, translated, retry_after) where status is + 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 for post in posts: try: translated += _translate_one(session, post, base_url, target) session.commit() - except ic.InterpreterUnavailable: + except ic.InterpreterUnavailable as e: session.rollback() - return "interrupted", translated + return "interrupted", translated, getattr(e, "retry_after", None) except ic.InterpreterBadRequest as e: session.rollback() log.warning("translate bad request: %s", e) - return "stopped", translated + return "stopped", translated, None except SoftTimeLimitExceeded: - return "timeout", translated - return "ok", translated + return "timeout", translated, None + return "ok", translated, None def _summary(status: str, translated: int, scanned: int) -> str: diff --git a/docker-compose.yml b/docker-compose.yml index ee793f7..60eeda1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -47,6 +47,14 @@ services: web: image: git.fabledsword.com/bvandeusen/fabledcurator:dev 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 ports: - "${PORT:-8080}:8080" environment: &app_env @@ -77,6 +85,8 @@ services: worker: image: git.fabledsword.com/bvandeusen/fabledcurator:dev command: ["worker"] + # Drain in-flight import/thumbnail/download tasks before SIGKILL on deploy. + stop_grace_period: 90s environment: <<: *app_env CELERY_QUEUES: default,import,thumbnail,download @@ -93,6 +103,8 @@ services: scheduler: image: git.fabledsword.com/bvandeusen/fabledcurator:dev command: ["scheduler"] + # Quick maintenance/scan lane + beat — short tasks, modest drain window. + stop_grace_period: 60s environment: <<: *app_env CELERY_QUEUES: maintenance,scan @@ -110,6 +122,10 @@ services: maintenance-long: image: git.fabledsword.com/bvandeusen/fabledcurator:dev 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 environment: <<: *app_env CELERY_QUEUES: maintenance_long @@ -125,6 +141,8 @@ services: ml-worker: image: git.fabledsword.com/bvandeusen/fabledcurator-ml:dev command: ["ml-worker"] + # A single GPU inference pass can run tens of seconds — let it finish. + stop_grace_period: 120s environment: <<: *app_env volumes: diff --git a/tests/test_interpreter_client.py b/tests/test_interpreter_client.py index 39b2e07..b84cdeb 100644 --- a/tests/test_interpreter_client.py +++ b/tests/test_interpreter_client.py @@ -6,9 +6,10 @@ from backend.app.services import interpreter_client as ic 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._payload = payload or {} + self.headers = headers or {} def json(self): return self._payload @@ -66,6 +67,38 @@ def test_translate_connection_error_raises_unavailable(monkeypatch): 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.requests, "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.requests, "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.requests, "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): monkeypatch.setattr( ic.requests, "post", lambda *a, **k: _Resp(400, {"error": "bad target"}) diff --git a/tests/test_translate_posts.py b/tests/test_translate_posts.py index 26f2c8b..4218c4b 100644 --- a/tests/test_translate_posts.py +++ b/tests/test_translate_posts.py @@ -4,6 +4,7 @@ import pytest from sqlalchemy import select 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 pytestmark = pytest.mark.integration @@ -239,3 +240,61 @@ def test_retranslate_unhealthy_does_not_reset(db_sync, monkeypatch): db_sync.refresh(p) assert p.translated_source_lang == "ja" 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