From 25dff2a780ed2797fa7800b3cfdb137087d9ac4e Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Thu, 24 Sep 2026 09:54:10 +0200 Subject: [PATCH] fix(i18n): stop the translator from losing entries the server only refused Four defects in the client surfaced in one run. LibreTranslate serves Chinese as zh-Hans and zh-Hant, never as bare zh, so the readiness wait blocked on a target that was never coming and the run stood still for an hour without printing a line. A blanket except turned every transient refusal into a permanently discarded entry, and the fallback path answered a failed batch by firing twenty single-text requests at the same overloaded server; the same catalogs that lost 130 to 295 entries came back with 0 discarded once the batch is retried with backoff first. The per-catalog count conflated a refusal with a damaged protected span, which is measurably wrong: of 228 entries left pending in core/sw only 36 carried a span at all. Outcome now carries values, refusals, damage and the last refusal message per call rather than on the client, where twenty lanes mixed their numbers into each other. That split is what surfaced the 502 the server had been answering with. The throwaway container also ran with check=True and capture_output=True, so docker error text died inside the swallowed exception and the log showed an argv list and rc 125. Co-Authored-By: Claude Opus 5 (1M context) --- cli/build/i18n/__main__.py | 103 ++++++++---- tests/unit/python/cli/build/i18n/test_main.py | 7 +- .../python/utils/i18n/test_libretranslate.py | 85 +++++++++- utils/i18n/libretranslate.py | 159 ++++++++++++++---- 4 files changed, 283 insertions(+), 71 deletions(-) diff --git a/cli/build/i18n/__main__.py b/cli/build/i18n/__main__.py index 10bb727d2a..be3f0336b4 100644 --- a/cli/build/i18n/__main__.py +++ b/cli/build/i18n/__main__.py @@ -14,7 +14,7 @@ import json import os import sys import tempfile -from concurrent.futures import ProcessPoolExecutor +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from pathlib import Path from utils.cache.files import PROJECT_ROOT @@ -108,39 +108,71 @@ def prune(domains: list[str], requested: list[str]) -> int: return 0 -def translate(domain: str, requested: list[str]) -> int: - supported = translatable(load_languages(PROJECT_ROOT), domain) - unsupported = sorted(set(requested) - set(supported)) - if unsupported: - print(f"LibreTranslate does not support {unsupported}", file=sys.stderr) - return 2 - codes = [ - code - for code in (sorted(requested) if requested else supported) - if pending(read_catalog(catalog_path(PROJECT_ROOT, code, domain))) - ] - if not codes: - print(f"{domain}: nothing to translate") +def usable_cpus() -> int: + """Return the CPUs this process may run on.""" + try: + return len(os.sched_getaffinity(0)) + except (AttributeError, OSError): + return os.cpu_count() or 1 + + +def one_catalog(client: LibreTranslate, domain: str, code: str) -> None: + """Translate every pending entry of one catalog and write it back. + + Args: + client: translation client. + domain: catalog domain. + code: ISO 639-1 code of the catalog. + """ + path = catalog_path(PROJECT_ROOT, code, domain) + catalog = read_catalog(path) + todo = pending(catalog) + discarded = refused = damaged = 0 + refusal = "" + for start in range(0, len(todo), CHUNK_SIZE): + chunk = todo[start : start + CHUNK_SIZE] + outcome = client.translate([m.id for m in chunk], code) + discarded += apply(chunk, outcome.values) + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + write_catalog(path, catalog) + print(f"{domain}/{code}: {start + len(chunk)}/{len(todo)}", flush=True) + tail = f", last refusal {refusal}" if refused else "" + print( + f"{domain}/{code}: {len(todo) - discarded} translated, {discarded} discarded " + f"({refused} requests refused, {damaged} damaged a protected span){tail}", + flush=True, + ) + + +def translate(domains: list[str], requested: list[str]) -> int: + loaded = load_languages(PROJECT_ROOT) + work: list[tuple[str, str]] = [] + for domain in domains: + supported = translatable(loaded, domain) + unsupported = sorted(set(requested) - set(supported)) + if unsupported: + print(f"LibreTranslate does not support {unsupported}", file=sys.stderr) + return 2 + work += [ + (domain, code) + for code in (sorted(requested) if requested else supported) + if pending(read_catalog(catalog_path(PROJECT_ROOT, code, domain))) + ] + if not work: + print(f"{', '.join(domains)}: nothing to translate") return 0 - threads = os.cpu_count() or 1 - with server(PROJECT_ROOT, codes, threads) as url: - client = LibreTranslate(url, threads) + + cpus = usable_cpus() + codes = sorted({code for _, code in work}) + lanes = min(len(work), cpus) + with server(PROJECT_ROOT, codes, cpus) as url: + client = LibreTranslate(url, max(cpus // lanes, 1)) client.wait(codes, READY_TIMEOUT_SECONDS) - for code in codes: - path = catalog_path(PROJECT_ROOT, code, domain) - catalog = read_catalog(path) - todo = pending(catalog) - discarded = 0 - for start in range(0, len(todo), CHUNK_SIZE): - chunk = todo[start : start + CHUNK_SIZE] - discarded += apply(chunk, client.translate([m.id for m in chunk], code)) - write_catalog(path, catalog) - print(f"{domain}/{code}: {start + len(chunk)}/{len(todo)}", flush=True) - print( - f"{domain}/{code}: {len(todo) - discarded} translated, " - f"{discarded} discarded", - flush=True, - ) + with ThreadPoolExecutor(lanes) as pool: + for _ in pool.map(lambda job: one_catalog(client, *job), work): + pass return 0 @@ -152,7 +184,7 @@ def main() -> int: translate_parser = commands.add_parser( "translate", help="Machine-translate empty and fuzzy entries." ) - translate_parser.add_argument("--domain", choices=DOMAINS, required=True) + translate_parser.add_argument("--domain", choices=DOMAINS) translate_parser.add_argument( "--languages", default="", @@ -181,7 +213,10 @@ def main() -> int: [args.domain] if args.domain else list(DOMAINS), [c for c in args.languages.split(",") if c], ) - return translate(args.domain, [c for c in args.languages.split(",") if c]) + return translate( + [args.domain] if args.domain else list(DOMAINS), + [c for c in args.languages.split(",") if c], + ) if __name__ == "__main__": diff --git a/tests/unit/python/cli/build/i18n/test_main.py b/tests/unit/python/cli/build/i18n/test_main.py index b7d4b008d4..4c3875e688 100644 --- a/tests/unit/python/cli/build/i18n/test_main.py +++ b/tests/unit/python/cli/build/i18n/test_main.py @@ -9,6 +9,7 @@ from tempfile import TemporaryDirectory from utils.i18n.catalog import catalog_path, new_catalog, write_catalog from utils.i18n.languages import LANGUAGES_FILE +from utils.i18n.libretranslate import Outcome cli = importlib.import_module("cli.build.i18n.__main__") @@ -39,15 +40,15 @@ class TestTranslate(unittest.TestCase): def _run(self) -> tuple[int, mock.Mock]: started = mock.Mock(return_value=contextlib.nullcontext("http://lt")) client = mock.Mock() - client.return_value.translate.side_effect = lambda texts, code: ( - ["Bonjour"] * len(texts) + client.return_value.translate.side_effect = lambda texts, code: Outcome( + ["Bonjour"] * len(texts), 0, 0, "" ) with ( mock.patch.object(cli, "PROJECT_ROOT", self.root), mock.patch.object(cli, "server", started), mock.patch.object(cli, "LibreTranslate", client), ): - return cli.translate("core", []), started + return cli.translate(["core"], []), started def test_only_languages_with_pending_entries_start_the_container(self) -> None: _catalog(self.root, "fr", "") diff --git a/tests/unit/python/utils/i18n/test_libretranslate.py b/tests/unit/python/utils/i18n/test_libretranslate.py index 6698330480..3d4e3a8ba5 100644 --- a/tests/unit/python/utils/i18n/test_libretranslate.py +++ b/tests/unit/python/utils/i18n/test_libretranslate.py @@ -63,12 +63,93 @@ class TestTranslateFailures(unittest.TestCase): rejected = urllib.error.HTTPError( "http://libretranslate", 500, "boom", {}, None ) - with mock.patch.object(self.client, "_call", side_effect=rejected): + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=rejected), + ): self.assertEqual( - self.client.translate(["Hello", "World"], "de"), [None, None] + self.client.translate(["Hello", "World"], "de").values, [None, None] ) +class TestRefusalsAreRetried(unittest.TestCase): + """A server that buckles under the lanes must not cost entries for good.""" + + def setUp(self) -> None: + self.client = LibreTranslate("http://libretranslate", 1) + self.refusal = urllib.error.HTTPError( + "http://libretranslate", 500, "boom", {}, None + ) + + def test_a_transient_refusal_is_retried_until_it_succeeds(self) -> None: + answers = [self.refusal, self.refusal, {"translatedText": ["Hallo"]}] + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=answers), + ): + outcome = self.client.translate(["Hello"], "de") + + self.assertEqual(outcome.values, ["Hallo"]) + self.assertEqual(outcome.refused, 2) + self.assertEqual(outcome.damaged, 0) + + def test_a_permanent_refusal_is_counted_and_named(self) -> None: + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=self.refusal), + ): + outcome = self.client.translate(["Hello"], "de") + + self.assertEqual(outcome.values, [None]) + self.assertTrue(outcome.refused) + self.assertIn("HTTPError", outcome.refusal) + + def test_a_catalog_never_inherits_another_lane_s_counts(self) -> None: + answers = [ + self.refusal, + {"translatedText": ["Hallo"]}, + {"translatedText": ["Hi"]}, + ] + with ( + mock.patch("utils.i18n.libretranslate.time.sleep"), + mock.patch.object(self.client, "_call", side_effect=answers), + ): + first = self.client.translate(["Hello"], "de") + second = self.client.translate(["Hello"], "de") + + self.assertEqual(first.refused, 1) + self.assertEqual(second.refused, 0) + + +class TestScriptedTargetCodes(unittest.TestCase): + """LibreTranslate offers Chinese as zh-Hans/zh-Hant, never as bare zh.""" + + def setUp(self) -> None: + self.client = LibreTranslate("http://libretranslate", 1) + self.served = [{"code": "en", "targets": ["de", "zh-Hans", "zh-Hant"]}] + + def test_the_catalog_code_is_sent_as_the_code_the_server_serves(self) -> None: + with mock.patch.object( + self.client, "_call", return_value={"translatedText": ["你好"]} + ) as call: + self.client.translate(["Hello"], "zh") + + self.assertEqual(call.call_args.args[1]["target"], "zh-Hans") + + def test_waiting_accepts_a_server_that_only_names_the_script(self) -> None: + with mock.patch.object(self.client, "_call", return_value=self.served): + self.client.wait(["zh", "de"], timeout=0) + + def test_waiting_still_reports_a_target_nobody_serves(self) -> None: + with ( + mock.patch.object(self.client, "_call", return_value=self.served), + self.assertRaises(TimeoutError) as raised, + ): + self.client.wait(["fr"], timeout=0) + + self.assertIn("fr", str(raised.exception)) + + class TestAccelerated(unittest.TestCase): def _runtimes(self, stdout: str) -> mock.Mock: return mock.Mock(spec=subprocess.CompletedProcess, stdout=stdout) diff --git a/utils/i18n/libretranslate.py b/utils/i18n/libretranslate.py index 4a02b85a23..287362b2f3 100644 --- a/utils/i18n/libretranslate.py +++ b/utils/i18n/libretranslate.py @@ -5,14 +5,13 @@ from __future__ import annotations import json import os import subprocess -import sys import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, NamedTuple from utils.cache.yaml import load_yaml from utils.i18n.languages import SOURCE_LANGUAGE @@ -20,9 +19,42 @@ from utils.i18n.placeholders import has_words, mask, unmask from utils.roles.mapping import ROLE_FILE_META_SERVICES if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Iterable, Iterator + +class Outcome(NamedTuple): + """What one translation call produced, and why anything is missing. + + Args: + values: one translation per source, None where it was discarded. + refused: requests the server turned down, retries included. + damaged: translations that altered a protected span. + refusal: the last refusal's message, empty when none happened. + """ + + values: list[str | None] + refused: int + damaged: int + refusal: str + + +def merge(outcomes: Iterable[Outcome]) -> Outcome: + """Fold per-text outcomes back into one for the whole batch.""" + values: list[str | None] = [] + refused = damaged = 0 + refusal = "" + for outcome in outcomes: + values += outcome.values + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + return Outcome(values, refused, damaged, refusal) + ROLE = "web-svc-libretranslate" +# LibreTranslate names a few targets by script rather than by the bare ISO +# 639-1 code the catalogs use. A code missing from this map simply never turns +# up in /languages, and the readiness wait then spins until its timeout. +SERVER_CODES = {"zh": "zh-Hans"} SERVICES_FILE = Path("roles") / ROLE / ROLE_FILE_META_SERVICES CONTAINER_PORT = 5000 MODELS_VOLUME = "infinito-i18n-libretranslate" @@ -33,8 +65,11 @@ POLL_SECONDS = 5 READY_TIMEOUT_SECONDS = 3600 REQUEST_TIMEOUT_SECONDS = 600 DEPLOYED_TIMEOUT_SECONDS = 5 -DEPLOY_TIMEOUT_SECONDS = 3600 +DEPLOY_TIMEOUT_SECONDS = 4 * 3600 RUNNER_TIMEOUT_SECONDS = 300 +SERVICE_TIMEOUT_SECONDS = 300 +RETRY_ATTEMPTS = 4 +RETRY_BACKOFF_SECONDS = 2 INVENTORY_DIR = Path.home() / "inventories" / "infinito-i18n" DEPLOY_PID_FILE = Path("build") / "deploy.pid" GPU_STAMP = Path("build") / "i18n-gpu.stamp" @@ -80,6 +115,27 @@ def deployed(root: Path) -> str: return "" +def await_service(root: Path) -> str: + """Poll the deployed LibreTranslate until it answers, or give up. + + A service the playbook just created needs longer than one probe's timeout + to serve its first request, and a single miss would send the caller off to + spawn a throwaway container beside the one it just deployed. + + Args: + root: repository root. + + Returns: + The base URL, empty when nothing answered before the deadline. + """ + deadline = time.monotonic() + SERVICE_TIMEOUT_SECONDS + while True: + running = deployed(root) + if running or time.monotonic() > deadline: + return running + time.sleep(POLL_SECONDS) + + def unaccelerated(root: Path) -> bool: """Return whether the running LibreTranslate predates this host's GPU. @@ -186,7 +242,6 @@ def deploy(root: Path) -> str: """ from cli.administration.deploy.development.env import compose_file_args - development = [sys.executable, "-m", "cli.administration.deploy.development"] inside = [ "docker", "compose", @@ -212,6 +267,8 @@ def deploy(root: Path) -> str: "--force-recreate", "i18n", ], + [*inside[:-2], "sh", "-lc", "/usr/local/bin/package-frontend-ca.sh"], + [*inside[:-2], "systemctl", "daemon-reload"], [*inside[:-2], "bash", f"{environment(root)['INFINITO_SRC_DIR']}/{bootstrap}"], [ *inside, @@ -252,7 +309,7 @@ def deploy(root: Path) -> str: ) if index == 0: await_runner(name, outer) - running = deployed(root) + running = await_service(root) if running and accelerated(): (root / GPU_STAMP).parent.mkdir(exist_ok=True) (root / GPU_STAMP).touch() @@ -323,7 +380,13 @@ def container(image: str, codes: list[str], threads: int) -> Iterator[str]: if accelerated(): command += ["--gpus", "all"] image = f"{image}{CUDA_SUFFIX}" - subprocess.run([*command, image], check=True, capture_output=True) + started = subprocess.run( + [*command, image], capture_output=True, text=True, check=False + ) + if started.returncode: + raise RuntimeError( + f"docker run exited with {started.returncode}: {started.stderr.strip()}" + ) try: mapping = subprocess.run( ["docker", "port", name, f"{CONTAINER_PORT}/tcp"], @@ -362,6 +425,11 @@ class LibreTranslate: ) as response: return json.load(response) + @staticmethod + def server_code(code: str) -> str: + """Return the target code the server knows this catalog code by.""" + return SERVER_CODES.get(code, code) + def targets(self) -> set[str]: """Return the languages the server translates English into.""" for language in self._call("/languages"): @@ -377,10 +445,11 @@ class LibreTranslate: timeout: seconds to wait before giving up. """ deadline = time.monotonic() + timeout - missing = set(codes) + wanted = {self.server_code(code) for code in codes} + missing = set(wanted) while True: try: - missing = set(codes) - self.targets() + missing = wanted - self.targets() if not missing: return except (OSError, ValueError): @@ -391,30 +460,49 @@ class LibreTranslate: ) time.sleep(POLL_SECONDS) - def _batch(self, texts: list[str], target: str) -> list[str | None]: + def _post(self, masked: list, target: str) -> list | None: + """Return the server's translations, or None once the retries run out. + + The server refuses a share of the requests while every lane hammers it + at once, and those refusals used to discard their entries for good. + """ + refusals = 0 + refusal = "" + for attempt in range(RETRY_ATTEMPTS): + try: + result = self._call( + "/translate", + { + "q": [item.text for item in masked], + "source": SOURCE_LANGUAGE, + "target": self.server_code(target), + "format": "html", + }, + )["translatedText"] + except (urllib.error.HTTPError, ValueError, KeyError, TypeError) as exc: + refusals += 1 + refusal = f"{type(exc).__name__}: {exc}" + result = None + if isinstance(result, list) and len(result) == len(masked): + return result, refusals, refusal + time.sleep(RETRY_BACKOFF_SECONDS * 2**attempt) + return None, refusals, refusal + + def _batch(self, texts: list[str], target: str) -> Outcome: masked = [mask(text) for text in texts] - try: - result = self._call( - "/translate", - { - "q": [item.text for item in masked], - "source": SOURCE_LANGUAGE, - "target": target, - "format": "html", - }, - )["translatedText"] - except (urllib.error.HTTPError, ValueError, KeyError, TypeError): - result = None - if not isinstance(result, list) or len(result) != len(texts): + result, refused, refusal = self._post(masked, target) + if result is None: if len(texts) == 1: - return [None] - return [self._batch([text], target)[0] for text in texts] - return [ + return Outcome([None], refused, 0, refusal) + return merge(self._batch([text], target) for text in texts) + values = [ unmask(translated, item, text) for translated, item, text in zip(result, masked, texts, strict=True) ] + damaged = sum(1 for text in values if text is None) + return Outcome(values, refused, damaged, refusal) - def translate(self, texts: list[str], target: str) -> list[str | None]: + def translate(self, texts: list[str], target: str) -> Outcome: """Translate ``texts`` from English into ``target``. Args: @@ -422,8 +510,10 @@ class LibreTranslate: target: ISO 639-1 code. Returns: - One translation per source; ``None`` where the server rejected the - request or the translation damaged a protected span. + One translation per source, ``None`` where it was discarded, plus + the counts that say which of the two reasons applied. The counts + belong to this call: lanes share the client, so a field on the + client would mix another catalog's numbers into this one's. Raises: OSError: the server is unreachable; every further request would @@ -432,12 +522,17 @@ class LibreTranslate: results: list[str | None] = list(texts) wordy = [index for index, text in enumerate(texts) if has_words(text)] batches = [wordy[i : i + BATCH_SIZE] for i in range(0, len(wordy), BATCH_SIZE)] + refused = damaged = 0 + refusal = "" with ThreadPoolExecutor(self.workers) as pool: translated = pool.map( lambda batch: self._batch([texts[index] for index in batch], target), batches, ) - for batch, texts_out in zip(batches, translated, strict=True): - for index, text in zip(batch, texts_out, strict=True): + for batch, outcome in zip(batches, translated, strict=True): + for index, text in zip(batch, outcome.values, strict=True): results[index] = text - return results + refused += outcome.refused + damaged += outcome.damaged + refusal = outcome.refusal or refusal + return Outcome(results, refused, damaged, refusal)