diff --git a/backend/app/services/interpreter_client.py b/backend/app/services/interpreter_client.py index aeb3a4e..c4db1a3 100644 --- a/backend/app/services/interpreter_client.py +++ b/backend/app/services/interpreter_client.py @@ -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"}, diff --git a/backend/app/tasks/translation.py b/backend/app/tasks/translation.py index 70f5b51..02cc153 100644 --- a/backend/app/tasks/translation.py +++ b/backend/app/tasks/translation.py @@ -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)) diff --git a/tests/test_interpreter_client.py b/tests/test_interpreter_client.py index b84cdeb..0c4adbc 100644 --- a/tests/test_interpreter_client.py +++ b/tests/test_interpreter_client.py @@ -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 diff --git a/tests/test_translate_posts.py b/tests/test_translate_posts.py index 4218c4b..2973241 100644 --- a/tests/test_translate_posts.py +++ b/tests/test_translate_posts.py @@ -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)