feat(ops): graceful shutdown — worker stop-grace + Interpreter drain resilience
CI / lint (push) Failing after 2s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 35s
CI / integration (push) Successful in 3m42s

Deploys (docker SIGTERM→SIGKILL, default 10s) were killing Celery jobs
mid-flight. Give in-flight work room to drain and make interrupted work
resume cleanly instead of stalling.

- docker-compose.yml: stop_grace_period per lane (web 30s / worker 90s /
  scheduler 60s / maintenance-long 180s / ml-worker 120s) so warm shutdown
  can actually drain before SIGKILL.
- celery_app.py: task_reject_on_worker_lost=True — a task killed past the
  grace window is re-queued (safe: idempotent + chunked, recovery sweeps
  re-drive stragglers).
- interpreter_client.py: map 429/5xx (502/503/504) → InterpreterUnavailable
  and parse Retry-After (delta-seconds or HTTP-date); a draining Interpreter
  behind a reverse proxy no longer raises an opaque HTTPError.
- translation.py: thread retry_after out of _translate_batch; retranslate_posts
  resumes after the Retry-After hint (or 60s default, capped 900s) on an
  interrupt with _reset_done=True, self-terminating via the health gate.
- tests: 429/5xx mapping + Retry-After parse; interrupt-resume + default backoff.

No migration.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-07 21:01:00 -04:00
parent 1f6d94f51d
commit c64261593d
6 changed files with 215 additions and 19 deletions
+8
View File
@@ -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
+48 -6
View File
@@ -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")
+48 -12
View File
@@ -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: