feat(translation): pooled Interpreter session + manual sweep resumes after drain
- interpreter_client: shared requests.Session with a connect-only retry (connect=2, no status retries — we map 429/5xx ourselves) so a proxy reload is smoothed and the keep-alive connection is pooled across the sweep. - translate_posts: on an interrupt (drain), re-enqueue after the Retry-After hint / default backoff instead of waiting for the daily beat; self-terminates via the health gate. Steady-state one-chunk-per-run on success is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -12,6 +12,8 @@ from datetime import UTC, datetime
|
||||
from email.utils import parsedate_to_datetime
|
||||
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util.retry import Retry
|
||||
|
||||
|
||||
class InterpreterUnavailable(Exception):
|
||||
@@ -57,13 +59,29 @@ def _parse_retry_after(resp) -> float | None:
|
||||
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:
|
||||
"""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."""
|
||||
if not base_url:
|
||||
return False
|
||||
try:
|
||||
r = requests.get(_url(base_url, "/v1/health"), timeout=timeout)
|
||||
r = session.get(_url(base_url, "/v1/health"), timeout=timeout)
|
||||
except requests.RequestException:
|
||||
return False
|
||||
if r.status_code != 200:
|
||||
@@ -94,7 +112,7 @@ def translate(
|
||||
return {"translations": [], "detected_lang": None,
|
||||
"engine": None, "engine_version": None}
|
||||
try:
|
||||
r = requests.post(
|
||||
r = session.post(
|
||||
_url(base_url, "/v1/translate"),
|
||||
json={"q": list(texts), "source": source,
|
||||
"target": target, "engine": "auto"},
|
||||
|
||||
@@ -70,10 +70,21 @@ def translate_posts() -> str:
|
||||
return ready
|
||||
base_url, target = ready
|
||||
posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN)
|
||||
# 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)
|
||||
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))
|
||||
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
|
||||
"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")
|
||||
assert out["translations"] == ["The cat is cute", "Blonde gal!"]
|
||||
assert out["detected_lang"] == "ja"
|
||||
@@ -41,7 +41,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
|
||||
|
||||
def test_translate_passthrough_unchanged(monkeypatch):
|
||||
# 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"],
|
||||
"detectedLanguage": {"language": "en"},
|
||||
"interpreter": {"engine": "none", "engine_version": None},
|
||||
@@ -53,7 +53,7 @@ def test_translate_passthrough_unchanged(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):
|
||||
ic.translate(["x"], base_url="http://i.lan")
|
||||
|
||||
@@ -62,7 +62,7 @@ def test_translate_connection_error_raises_unavailable(monkeypatch):
|
||||
def boom(*a, **k):
|
||||
raise ic.requests.ConnectionError("refused")
|
||||
|
||||
monkeypatch.setattr(ic.requests, "post", boom)
|
||||
monkeypatch.setattr(ic.session, "post", boom)
|
||||
with pytest.raises(ic.InterpreterUnavailable):
|
||||
ic.translate(["x"], base_url="http://i.lan")
|
||||
|
||||
@@ -71,7 +71,7 @@ def test_translate_connection_error_raises_unavailable(monkeypatch):
|
||||
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, {}))
|
||||
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
|
||||
@@ -79,7 +79,7 @@ def test_translate_429_and_5xx_raise_unavailable(monkeypatch, code):
|
||||
|
||||
def test_translate_honours_retry_after_seconds(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
ic.requests, "post",
|
||||
ic.session, "post",
|
||||
lambda *a, **k: _Resp(503, {}, headers={"Retry-After": "42"}),
|
||||
)
|
||||
with pytest.raises(ic.InterpreterUnavailable) as ei:
|
||||
@@ -90,7 +90,7 @@ def test_translate_honours_retry_after_seconds(monkeypatch):
|
||||
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",
|
||||
ic.session, "post",
|
||||
lambda *a, **k: _Resp(
|
||||
503, {}, headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}),
|
||||
)
|
||||
@@ -101,7 +101,7 @@ def test_translate_retry_after_past_http_date_clamps_to_zero(monkeypatch):
|
||||
|
||||
def test_translate_400_raises_bad_request(monkeypatch):
|
||||
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):
|
||||
ic.translate(["x"], base_url="http://i.lan")
|
||||
@@ -111,20 +111,20 @@ def test_translate_empty_is_noop(monkeypatch):
|
||||
def boom(*a, **k):
|
||||
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"] == []
|
||||
|
||||
|
||||
def test_health_true_when_llm_up(monkeypatch):
|
||||
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
|
||||
|
||||
|
||||
def test_health_false_when_engine_down(monkeypatch):
|
||||
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
|
||||
|
||||
@@ -133,7 +133,7 @@ def test_health_false_on_error(monkeypatch):
|
||||
def boom(*a, **k):
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -137,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
|
||||
|
||||
|
||||
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"):
|
||||
"""Interpreter is up and translates with a NEW engine version."""
|
||||
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
|
||||
|
||||
Reference in New Issue
Block a user