From 0332c94d1fb74a1413bcad6ccacefb3cd6b3746b Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Sat, 10 Oct 2026 14:17:00 -0700 Subject: [PATCH 1/2] fix(gt-i18n): keep translations loaded after the cache expires t() reads the cache through get_translations_sync, which returned {} once an entry was older than 60 s, and nothing reloaded it. Every locale went back to the source text a minute after startup. The cache now serves the last loaded translations after expiry while one reload runs on a daemon thread, and keeps them when that reload fails. Only a locale's first load makes callers wait. I18nManager gains get_translations_blocking and load_all_translations_blocking for synchronous code. The CDN loader defaults to https://cdn.gtx.dev, appends version_id and branch_id like the JS SDK, and treats a failed or empty response as a failed load, because the CDN answers backend errors with 200 {}. Co-Authored-By: Claude Opus 5.5 --- .sampo/changesets/keep-translations-loaded.md | 5 + packages/gt-i18n/pyproject.toml | 4 +- .../src/gt_i18n/i18n_manager/_i18n_manager.py | 22 +- .../gt_i18n/i18n_manager/_remote_loader.py | 36 ++- .../i18n_manager/_translations_manager.py | 166 +++++++++-- .../src/gt_i18n/internal/_load_gt_config.py | 4 +- packages/gt-i18n/tests/test_remote_loader.py | 75 +++++ .../tests/test_translations_manager.py | 263 ++++++++++++++++++ 8 files changed, 527 insertions(+), 48 deletions(-) create mode 100644 .sampo/changesets/keep-translations-loaded.md create mode 100644 packages/gt-i18n/tests/test_remote_loader.py diff --git a/.sampo/changesets/keep-translations-loaded.md b/.sampo/changesets/keep-translations-loaded.md new file mode 100644 index 0000000..ab13b19 --- /dev/null +++ b/.sampo/changesets/keep-translations-loaded.md @@ -0,0 +1,5 @@ +--- +pypi/gt-i18n: patch +--- + +fix: keep translations loaded after the 60 s cache expiry (serve the cached copy while one background reload runs, keep it when a reload fails), default the CDN URL to cdn.gtx.dev and add the pinned `version_id` and `branch_id` to CDN requests, treat an empty `cache_url` as no CDN loading, and require generaltranslation 0.1.0 and generaltranslation-supported-locales 0.0.6 or later diff --git a/packages/gt-i18n/pyproject.toml b/packages/gt-i18n/pyproject.toml index 5bdaf9e..6a76e61 100644 --- a/packages/gt-i18n/pyproject.toml +++ b/packages/gt-i18n/pyproject.toml @@ -10,8 +10,8 @@ requires-python = ">=3.10" license = "FSL-1.1-ALv2" license-files = ["LICENSE.md"] dependencies = [ - "generaltranslation", - "generaltranslation-supported-locales", + "generaltranslation>=0.1.0", + "generaltranslation-supported-locales>=0.0.6", "httpx>=0.25.0", ] diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py index ea6a001..076da32 100644 --- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py +++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py @@ -24,12 +24,15 @@ class I18nManager: default_locale: The source/default locale. locales: List of target locales the application supports. project_id: GT project ID (used for CDN loader). - cache_url: CDN base URL override. + cache_url: CDN base URL override. ``None`` uses the General + Translation CDN; an empty string turns CDN loading off. store_adapter: Custom storage adapter. Defaults to :class:`ContextVarStorageAdapter`. load_translations: Custom translation loader. Overrides the remote CDN loader when provided. cache_expiry_time: Cache expiry in milliseconds. + version_id: Pinned translation version for CDN loading. + branch_id: General Translation branch of the pinned version. """ def __init__( @@ -44,6 +47,7 @@ def __init__( load_translations: TranslationsLoader | None = None, cache_expiry_time: int = 60_000, version_id: str | None = None, + branch_id: str | None = None, ) -> None: self._version_id = version_id self._default_locale = default_locale @@ -59,12 +63,12 @@ def __init__( # Translation loading if load_translations is not None: loader = load_translations - elif project_id: + elif project_id and cache_url != "": from gt_i18n.i18n_manager._remote_loader import ( create_remote_translation_loader, ) - loader = create_remote_translation_loader(project_id, cache_url or "") + loader = create_remote_translation_loader(project_id, cache_url, version_id=version_id, branch_id=branch_id) else: loader = lambda locale: {} # noqa: E731 @@ -114,8 +118,13 @@ async def get_translations(self, locale: str | None = None) -> dict[str, str]: target = locale or self.get_locale() return await self._translations.get_translations(target) + def get_translations_blocking(self, locale: str | None = None) -> dict[str, str]: + """Get translations for a locale (sync; blocks on the first load, reloads expired ones in the background).""" + target = locale or self.get_locale() + return self._translations.get_translations_blocking(target) + def get_translations_sync(self, locale: str | None = None) -> dict[str, str]: - """Get cached translations (sync, returns empty if not loaded).""" + """Get the last loaded translations (sync, never loads; empty if never loaded).""" target = locale or self.get_locale() return self._translations.get_translations_sync(target) @@ -123,3 +132,8 @@ async def load_all_translations(self) -> None: """Eagerly load translations for all configured locales.""" if self._locales: await self._translations.load_all(self._locales) + + def load_all_translations_blocking(self) -> None: + """Eagerly load translations for all configured locales from synchronous code.""" + if self._locales: + self._translations.load_all_blocking(self._locales) diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py index f8264a0..8f372e1 100644 --- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py +++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py @@ -12,29 +12,37 @@ def create_remote_translation_loader( project_id: str, - cache_url: str = DEFAULT_CACHE_URL, + cache_url: str | None = None, + *, + version_id: str | None = None, + branch_id: str | None = None, ) -> TranslationsLoader: """Create a loader that fetches translations from the CDN. Args: project_id: The GT project ID. - cache_url: CDN base URL. + cache_url: CDN base URL. Defaults to the General Translation CDN. + version_id: Pins the translations to one published version. + branch_id: The General Translation branch the version belongs to. Returns: - An async callable ``(locale) -> dict[str, str]``. + An async callable ``(locale) -> dict[str, str]`` that raises when the + request fails or returns no translations, so the cache keeps its + previous translations. The CDN answers some server-side failures with + an empty object, which is why empty counts as a failure here. """ + base = f"{cache_url or DEFAULT_CACHE_URL}/{project_id}" + version = f"/{version_id}" if version_id else "" + params = {"branchId": branch_id} if branch_id else None async def loader(locale: str) -> dict[str, str]: - url = f"{cache_url}/{project_id}/{locale}" - try: - async with httpx.AsyncClient() as client: - resp = await client.get(url, timeout=10.0) - if resp.status_code == 200: - data = resp.json() - if isinstance(data, dict): - return data - except Exception: - pass - return {} + url = f"{base}/{locale}{version}" + async with httpx.AsyncClient() as client: + resp = await client.get(url, params=params, timeout=10.0) + resp.raise_for_status() + data = resp.json() + if not isinstance(data, dict) or not data: + raise ValueError(f"No translations in the response from {url}") + return data return loader diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py index 7b687cd..99250ee 100644 --- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py +++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py @@ -4,8 +4,11 @@ import asyncio import inspect +import threading import time from collections.abc import Awaitable, Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any TranslationsLoader = Callable[[str], dict[str, str] | Awaitable[dict[str, str]]] @@ -18,8 +21,30 @@ def __init__(self, translations: dict[str, str], loaded_at: float) -> None: self.loaded_at = loaded_at +async def _await(awaitable: Awaitable[Any]) -> Any: + return await awaitable + + +def _run_blocking(awaitable: Awaitable[Any]) -> Any: + """Run *awaitable* to completion from synchronous code.""" + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(_await(awaitable)) + # A loop is already running in this thread, so run it on a fresh one in another thread. + with ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(asyncio.run, _await(awaitable)).result() + + class TranslationsManager: - """Caches ``dict[str, str]`` (hash -> translated string) per locale.""" + """Caches ``dict[str, str]`` (hash -> translated string) per locale. + + An entry expires ``cache_expiry_time`` ms after it loads. An expired entry + keeps being served while one reload runs in the background, and a reload + that fails keeps the previous translations, so a slow or failing loader + never drops a loaded locale back to the source text. Only a locale's first + load makes its callers wait. + """ def __init__( self, @@ -29,47 +54,134 @@ def __init__( self._loader = loader self._cache_expiry_ms = cache_expiry_time self._cache: dict[str, _CacheEntry] = {} - self._loading: dict[str, asyncio.Task[dict[str, str]]] = {} + self._loading: dict[str, asyncio.Task[None]] = {} + self._locks: dict[str, threading.Lock] = {} + self._locks_guard = threading.Lock() def _is_expired(self, entry: _CacheEntry) -> bool: elapsed_ms = (time.monotonic() - entry.loaded_at) * 1000 return elapsed_ms > self._cache_expiry_ms - async def _call_loader(self, locale: str) -> dict[str, str]: - result = self._loader(locale) - if inspect.isawaitable(result): - return await result - return result # type: ignore[return-value] - - async def get_translations(self, locale: str) -> dict[str, str]: - """Load translations for *locale*, using cache when valid.""" + def _fresh(self, locale: str) -> dict[str, str] | None: entry = self._cache.get(locale) if entry is not None and not self._is_expired(entry): return entry.translations + return None - # Deduplicate concurrent loads - if locale in self._loading: - return await self._loading[locale] + def _store(self, locale: str, translations: dict[str, str] | None) -> dict[str, str]: + """Cache a load result; ``None`` is a failed load, which keeps the previous translations.""" + if translations is None: + previous = self._cache.get(locale) + translations = previous.translations if previous is not None else {} + self._cache[locale] = _CacheEntry(translations, time.monotonic()) + return translations - task = asyncio.ensure_future(self._call_loader(locale)) - self._loading[locale] = task + # Async path (FastAPI and other asyncio code) + + async def _load_async(self, locale: str) -> dict[str, str] | None: try: - translations = await task + if inspect.iscoroutinefunction(self._loader): + return await self._loader(locale) + # A synchronous loader runs on a worker thread, off the event loop. + result = await asyncio.to_thread(self._loader, locale) + return await result if inspect.isawaitable(result) else result except Exception: - translations = {} - finally: - self._loading.pop(locale, None) + return None - self._cache[locale] = _CacheEntry(translations, time.monotonic()) - return translations + async def _load_and_store(self, locale: str) -> None: + self._store(locale, await self._load_async(locale)) - def get_translations_sync(self, locale: str) -> dict[str, str]: - """Return cached translations or empty dict (never blocks).""" + def _start_load(self, locale: str) -> asyncio.Task[None]: + """Start a load for *locale* on the running loop, or join the one already running there.""" + loop = asyncio.get_running_loop() + task = self._loading.get(locale) + if task is None or task.done() or task.get_loop() is not loop: + task = loop.create_task(self._load_and_store(locale)) + self._loading[locale] = task + task.add_done_callback(lambda done: self._forget_load(locale, done)) + return task + + def _forget_load(self, locale: str, task: asyncio.Task[None]) -> None: + if self._loading.get(locale) is task: + del self._loading[locale] + + async def get_translations(self, locale: str) -> dict[str, str]: + """Return translations for *locale*, loading them if missing and reloading in the background if expired.""" + fresh = self._fresh(locale) + if fresh is not None: + return fresh entry = self._cache.get(locale) - if entry is not None and not self._is_expired(entry): + if entry is not None: + # The reload runs on a thread, so it outlives this call's event loop. + self._refresh_in_background(locale) return entry.translations - return {} + # Shielded, so a cancelled caller does not cancel the load for the others. + await asyncio.shield(self._start_load(locale)) + return self.get_translations_sync(locale) async def load_all(self, locales: list[str]) -> None: - """Eagerly fetch translations for all *locales*.""" - await asyncio.gather(*(self.get_translations(loc) for loc in locales)) + """Eagerly fetch translations for all *locales*, waiting for every load.""" + tasks = [self._start_load(loc) for loc in locales if self._fresh(loc) is None] + if tasks: + await asyncio.gather(*(asyncio.shield(task) for task in tasks)) + + # Blocking path (Flask and other synchronous code) + + def _lock_for(self, locale: str) -> threading.Lock: + with self._locks_guard: + return self._locks.setdefault(locale, threading.Lock()) + + def _load_blocking(self, locale: str) -> dict[str, str] | None: + try: + result = self._loader(locale) + return _run_blocking(result) if inspect.isawaitable(result) else result + except Exception: + return None + + def _reload_in_background(self, locale: str, lock: threading.Lock) -> None: + try: + self._store(locale, self._load_blocking(locale)) + finally: + lock.release() + + def _refresh_in_background(self, locale: str) -> None: + """Reload an expired *locale* on a daemon thread, unless a reload is already running.""" + lock = self._lock_for(locale) + if not lock.acquire(blocking=False): + return + if self._fresh(locale) is not None: # another reload finished in the meantime + lock.release() + return + try: + threading.Thread(target=self._reload_in_background, args=(locale, lock), daemon=True).start() + except BaseException: + lock.release() + raise + + def get_translations_blocking(self, locale: str) -> dict[str, str]: + """Return translations for *locale* from synchronous code. + + The first load blocks the callers for that locale. After that, an + expired entry is returned at once while one background thread reloads it. + """ + fresh = self._fresh(locale) + if fresh is not None: + return fresh + entry = self._cache.get(locale) + if entry is not None: + self._refresh_in_background(locale) + return entry.translations + with self._lock_for(locale): + entry = self._cache.get(locale) + if entry is not None: + return entry.translations + return self._store(locale, self._load_blocking(locale)) + + def load_all_blocking(self, locales: list[str]) -> None: + """Eagerly fetch translations for all *locales* from synchronous code.""" + _run_blocking(self.load_all(locales)) + + def get_translations_sync(self, locale: str) -> dict[str, str]: + """Return the last loaded translations, even if expired, or an empty dict (never blocks).""" + entry = self._cache.get(locale) + return entry.translations if entry is not None else {} diff --git a/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py b/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py index 4c088df..0f9f4ec 100644 --- a/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py +++ b/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py @@ -17,6 +17,7 @@ "runtimeUrl": "runtime_url", "customMapping": "custom_mapping", "_versionId": "version_id", + "_branchId": "branch_id", } @@ -24,10 +25,11 @@ class GTConfig(TypedDict, total=False): default_locale: str locales: list[str] project_id: str - cache_url: str + cache_url: str | None runtime_url: str custom_mapping: CustomMapping version_id: str + branch_id: str def load_gt_config(config_path: str | None = None) -> GTConfig: diff --git a/packages/gt-i18n/tests/test_remote_loader.py b/packages/gt-i18n/tests/test_remote_loader.py new file mode 100644 index 0000000..bddae25 --- /dev/null +++ b/packages/gt-i18n/tests/test_remote_loader.py @@ -0,0 +1,75 @@ +"""Tests for the CDN translation loader.""" + +import asyncio +from typing import Any + +import httpx +import pytest +from gt_i18n import I18nManager +from gt_i18n.i18n_manager._remote_loader import create_remote_translation_loader + + +@pytest.fixture +def requests(monkeypatch: pytest.MonkeyPatch) -> list[httpx.Request]: + """Route the loader's HTTP calls to a fake CDN and record them.""" + import gt_i18n.i18n_manager._remote_loader as mod + + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.url.path.endswith("/missing"): + return httpx.Response(404) + if request.url.path.endswith("/empty"): + return httpx.Response(200, json={}) # what the CDN returns when its backend fails + return httpx.Response(200, json={"h": "Hola"}) + + real_client = httpx.AsyncClient + + def client(**kwargs: Any) -> httpx.AsyncClient: + return real_client(transport=httpx.MockTransport(handler), **kwargs) + + monkeypatch.setattr(mod.httpx, "AsyncClient", client) + return seen + + +def test_defaults_to_the_gt_cdn(requests: list[httpx.Request]) -> None: + loader = create_remote_translation_loader("proj") + assert asyncio.run(loader("es")) == {"h": "Hola"} # type: ignore[arg-type] + assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es" + + +def test_version_id_pins_the_url(requests: list[httpx.Request]) -> None: + loader = create_remote_translation_loader("proj", "https://cdn.example", version_id="v42") + asyncio.run(loader("es")) # type: ignore[arg-type] + assert str(requests[0].url) == "https://cdn.example/proj/es/v42" + + +def test_failed_request_raises(requests: list[httpx.Request]) -> None: + loader = create_remote_translation_loader("proj") + with pytest.raises(httpx.HTTPStatusError): + asyncio.run(loader("missing")) # type: ignore[arg-type] + + +def test_manager_without_cache_url_uses_the_gt_cdn(requests: list[httpx.Request]) -> None: + manager = I18nManager(default_locale="en", locales=["es"], project_id="proj", version_id="v7") + assert asyncio.run(manager.get_translations("es")) == {"h": "Hola"} + assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es/v7" + + +def test_empty_response_raises(requests: list[httpx.Request]) -> None: + loader = create_remote_translation_loader("proj") + with pytest.raises(ValueError): + asyncio.run(loader("empty")) # type: ignore[arg-type] + + +def test_branch_id_is_sent_as_a_query(requests: list[httpx.Request]) -> None: + loader = create_remote_translation_loader("proj", version_id="v42", branch_id="br_7") + asyncio.run(loader("es")) # type: ignore[arg-type] + assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es/v42?branchId=br_7" + + +def test_empty_cache_url_turns_cdn_loading_off(requests: list[httpx.Request]) -> None: + manager = I18nManager(default_locale="en", locales=["es"], project_id="proj", cache_url="") + assert asyncio.run(manager.get_translations("es")) == {} + assert requests == [] diff --git a/packages/gt-i18n/tests/test_translations_manager.py b/packages/gt-i18n/tests/test_translations_manager.py index 36dff79..4bc6f2e 100644 --- a/packages/gt-i18n/tests/test_translations_manager.py +++ b/packages/gt-i18n/tests/test_translations_manager.py @@ -1,6 +1,9 @@ """Tests for TranslationsManager.""" import asyncio +import threading +from collections.abc import Callable +from types import SimpleNamespace import pytest from gt_i18n.i18n_manager._translations_manager import TranslationsManager @@ -76,3 +79,263 @@ async def loader(locale: str) -> dict[str, str]: mgr = TranslationsManager(loader) result = asyncio.run(mgr.get_translations("es")) assert result == {"async_hash": "async_value"} + + +class _Clock: + """Stands in for ``time.monotonic`` in the cache module only, so asyncio's own clock keeps running.""" + + def __init__(self) -> None: + self.now = 1_000.0 + + def __call__(self) -> float: + return self.now + + +@pytest.fixture +def clock(monkeypatch: pytest.MonkeyPatch) -> _Clock: + import gt_i18n.i18n_manager._translations_manager as mod + + c = _Clock() + monkeypatch.setattr(mod, "time", SimpleNamespace(monotonic=c)) + return c + + +def _counting(values: list[dict[str, str]]) -> tuple[Callable[[str], dict[str, str]], list[int]]: + """A loader that returns *values* in order (repeating the last) and counts its calls.""" + calls = [0] + + def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + return values[min(calls[0], len(values)) - 1] + + return loader, calls + + +def _settle(mgr: TranslationsManager, locale: str) -> None: + """Wait for a background reload of *locale* to finish.""" + with mgr._lock_for(locale): + pass + + +def test_sync_serves_last_loaded_after_expiry(clock: _Clock, sample_translations: dict[str, str]) -> None: + mgr = TranslationsManager(lambda locale: sample_translations, cache_expiry_time=60_000) + asyncio.run(mgr.get_translations("es")) + clock.now += 61 + assert mgr.get_translations_sync("es") == sample_translations + + +def test_expired_entry_is_served_while_it_reloads(clock: _Clock) -> None: + loader, calls = _counting([{"h": "v1"}, {"h": "v2"}]) + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + + async def scenario() -> tuple[dict[str, str], dict[str, str]]: + await mgr.get_translations("es") + clock.now += 61 + served = await mgr.get_translations("es") # the stale copy, at once + _settle(mgr, "es") + return served, mgr.get_translations_sync("es") + + assert asyncio.run(scenario()) == ({"h": "v1"}, {"h": "v2"}) + assert calls[0] == 2 + + +def test_one_off_event_loops_still_refresh_after_expiry(clock: _Clock) -> None: + loader, calls = _counting([{"h": "v1"}, {"h": "v2"}]) + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"} + clock.now += 61 + assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"} # this loop ends right away + _settle(mgr, "es") + assert mgr.get_translations_sync("es") == {"h": "v2"} + assert calls[0] == 2 + + +def test_failed_reload_keeps_previous_translations(clock: _Clock, sample_translations: dict[str, str]) -> None: + calls = [0] + + def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + if calls[0] > 1: + raise RuntimeError("CDN unavailable") + return sample_translations + + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + + async def scenario() -> None: + await mgr.get_translations("es") + clock.now += 61 + await mgr.get_translations("es") + _settle(mgr, "es") + # The failed attempt counts as a load, so nothing retries until the next expiry. + await mgr.get_translations("es") + + asyncio.run(scenario()) + assert mgr.get_translations_sync("es") == sample_translations + assert calls[0] == 2 + + +def test_concurrent_first_loads_share_one_load() -> None: + calls = [0] + + async def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + await asyncio.sleep(0.01) + return {"h": locale} + + mgr = TranslationsManager(loader) + + async def scenario() -> list[dict[str, str]]: + return list(await asyncio.gather(*(mgr.get_translations("es") for _ in range(5)))) + + assert asyncio.run(scenario()) == [{"h": "es"}] * 5 + assert calls[0] == 1 + + +def test_cancelled_caller_does_not_cancel_the_load() -> None: + async def loader(locale: str) -> dict[str, str]: + await asyncio.sleep(0.05) + return {"h": locale} + + mgr = TranslationsManager(loader) + + async def scenario() -> dict[str, str]: + first = asyncio.ensure_future(mgr.get_translations("es")) + second = asyncio.ensure_future(mgr.get_translations("es")) + await asyncio.sleep(0.01) + first.cancel() + return await second + + assert asyncio.run(scenario()) == {"h": "es"} + assert mgr.get_translations_sync("es") == {"h": "es"} + + +def test_separate_event_loops_do_not_share_loads() -> None: + async def loader(locale: str) -> dict[str, str]: + await asyncio.sleep(0.05) + return {"h": locale} + + mgr = TranslationsManager(loader) + results: list[dict[str, str]] = [] + errors: list[BaseException] = [] + + def worker() -> None: + try: + results.append(asyncio.run(mgr.get_translations("es"))) + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(2)] + for th in threads: + th.start() + for th in threads: + th.join() + assert errors == [] + assert results == [{"h": "es"}, {"h": "es"}] + + +def test_sync_loader_runs_off_the_event_loop() -> None: + loader_threads: list[int] = [] + + def loader(locale: str) -> dict[str, str]: + loader_threads.append(threading.get_ident()) + return {"h": locale} + + mgr = TranslationsManager(loader) + + async def scenario() -> int: + await mgr.get_translations("es") + return threading.get_ident() + + loop_thread = asyncio.run(scenario()) + assert loader_threads and loader_threads[0] != loop_thread + + +def test_blocking_first_load_waits_and_loads_once() -> None: + gate = threading.Event() + calls = [0] + + def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + gate.wait(5) + return {"h": locale} + + mgr = TranslationsManager(loader) + results: list[dict[str, str]] = [] + threads = [threading.Thread(target=lambda: results.append(mgr.get_translations_blocking("es"))) for _ in range(4)] + for th in threads: + th.start() + gate.set() + for th in threads: + th.join() + assert results == [{"h": "es"}] * 4 + assert calls[0] == 1 + + +def test_blocking_expired_entry_is_served_without_waiting(clock: _Clock) -> None: + gate = threading.Event() + calls = [0] + + def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + if calls[0] > 1: + gate.wait(5) # a slow reload + return {"h": f"v{calls[0]}"} + + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + assert mgr.get_translations_blocking("es") == {"h": "v1"} + clock.now += 61 + assert mgr.get_translations_blocking("es") == {"h": "v1"} # returns while the reload is blocked + assert mgr.get_translations_blocking("es") == {"h": "v1"} # and does not start a second reload + gate.set() + _settle(mgr, "es") + assert mgr.get_translations_sync("es") == {"h": "v2"} + assert calls[0] == 2 + + +def test_blocking_failed_reload_keeps_previous_translations(clock: _Clock) -> None: + calls = [0] + + def loader(locale: str) -> dict[str, str]: + calls[0] += 1 + if calls[0] > 1: + raise RuntimeError("CDN unavailable") + return {"h": "v1"} + + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + mgr.get_translations_blocking("es") + clock.now += 61 + mgr.get_translations_blocking("es") + _settle(mgr, "es") + assert mgr.get_translations_sync("es") == {"h": "v1"} + + +def test_blocking_load_runs_async_loader() -> None: + async def loader(locale: str) -> dict[str, str]: + return {"async_hash": locale} + + mgr = TranslationsManager(loader) + assert mgr.get_translations_blocking("fr") == {"async_hash": "fr"} + + +def test_blocking_load_inside_running_loop() -> None: + async def loader(locale: str) -> dict[str, str]: + return {"h": locale} + + mgr = TranslationsManager(loader) + + async def caller() -> dict[str, str]: + return mgr.get_translations_blocking("de") + + assert asyncio.run(caller()) == {"h": "de"} + + +def test_load_all_blocking_inside_running_loop() -> None: + loader, calls = _counting([{"h": "x"}]) + mgr = TranslationsManager(loader) + + async def caller() -> None: + mgr.load_all_blocking(["es", "fr"]) + + asyncio.run(caller()) + assert calls[0] == 2 + assert mgr.get_translations_sync("fr") == {"h": "x"} From 3b065a2be835dde8bafc37bbbde6b068e3067ef7 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Sat, 17 Oct 2026 00:12:14 -0700 Subject: [PATCH 2/2] fix(gt-i18n): reload async loaders on their own event loop An expired read reloaded every locale on a daemon thread with a new event loop. An async loader that holds a client bound to the app's loop loaded once and then failed on every reload, and each failure kept the old translations, so they never updated. load_all() and that reload also used separate controls, so both could load a locale at the same time and the slower response replaced the newer translations. An async loader called from async code now reloads on the caller's loop, in the task that load_all() and first loads on that loop join. If the loop closes before the reload finishes, as a one-off asyncio.run() does, the reload finishes on a daemon thread. Every load that runs on a thread holds the locale's lock: a synchronous loader's loads from async code, load_all_blocking() and the background reloads. An expired async read starts no reload while a thread holds that lock. Co-Authored-By: Claude Opus 5.5 --- .../i18n_manager/_translations_manager.py | 54 +++++++--- .../tests/test_translations_manager.py | 101 +++++++++++++++++- 2 files changed, 142 insertions(+), 13 deletions(-) diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py index 99250ee..4ad0e7c 100644 --- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py +++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py @@ -36,6 +36,11 @@ def _run_blocking(awaitable: Awaitable[Any]) -> Any: return pool.submit(asyncio.run, _await(awaitable)).result() +def _is_async(loader: TranslationsLoader) -> bool: + """Whether *loader* is an ``async def`` function or an object whose ``__call__`` is one.""" + return inspect.iscoroutinefunction(loader) or inspect.iscoroutinefunction(getattr(loader, "__call__", None)) + + class TranslationsManager: """Caches ``dict[str, str]`` (hash -> translated string) per locale. @@ -44,6 +49,13 @@ class TranslationsManager: that fails keeps the previous translations, so a slow or failing loader never drops a loaded locale back to the source text. Only a locale's first load makes its callers wait. + + From async code, an async loader runs on the caller's event loop, so a + loader that holds a client bound to that loop keeps working, and its loads + of a locale on that loop share one task. If the loop closes before a load + finishes, as a one-off ``asyncio.run()`` does, the load finishes on a + thread. Every other load runs on a thread and holds the locale's lock. In + both cases, a reload and ``load_all`` for the same locale share one load. """ def __init__( @@ -52,6 +64,7 @@ def __init__( cache_expiry_time: int = 60_000, ) -> None: self._loader = loader + self._loader_is_async = _is_async(loader) self._cache_expiry_ms = cache_expiry_time self._cache: dict[str, _CacheEntry] = {} self._loading: dict[str, asyncio.Task[None]] = {} @@ -80,16 +93,17 @@ def _store(self, locale: str, translations: dict[str, str] | None) -> dict[str, async def _load_async(self, locale: str) -> dict[str, str] | None: try: - if inspect.iscoroutinefunction(self._loader): - return await self._loader(locale) - # A synchronous loader runs on a worker thread, off the event loop. - result = await asyncio.to_thread(self._loader, locale) + result = self._loader(locale) return await result if inspect.isawaitable(result) else result except Exception: return None async def _load_and_store(self, locale: str) -> None: - self._store(locale, await self._load_async(locale)) + if self._loader_is_async: + self._store(locale, await self._load_async(locale)) + else: + # A synchronous loader runs on a worker thread, off the event loop. + await asyncio.to_thread(self._load_locked, locale) def _start_load(self, locale: str) -> asyncio.Task[None]: """Start a load for *locale* on the running loop, or join the one already running there.""" @@ -98,12 +112,15 @@ def _start_load(self, locale: str) -> asyncio.Task[None]: if task is None or task.done() or task.get_loop() is not loop: task = loop.create_task(self._load_and_store(locale)) self._loading[locale] = task - task.add_done_callback(lambda done: self._forget_load(locale, done)) + task.add_done_callback(lambda done: self._load_finished(locale, done)) return task - def _forget_load(self, locale: str, task: asyncio.Task[None]) -> None: + def _load_finished(self, locale: str, task: asyncio.Task[None]) -> None: if self._loading.get(locale) is task: del self._loading[locale] + if task.cancelled(): + # The loop closed before the load finished, so it finishes on a thread. + self._refresh_in_background(locale) async def get_translations(self, locale: str) -> dict[str, str]: """Return translations for *locale*, loading them if missing and reloading in the background if expired.""" @@ -112,8 +129,12 @@ async def get_translations(self, locale: str) -> dict[str, str]: return fresh entry = self._cache.get(locale) if entry is not None: - # The reload runs on a thread, so it outlives this call's event loop. - self._refresh_in_background(locale) + if not self._loader_is_async: + # The reload runs on a thread, so it outlives this call's event loop. + self._refresh_in_background(locale) + elif not self._lock_for(locale).locked(): # unless a thread is reloading it already + # An async loader reloads on this loop, which it may hold a client for. + self._start_load(locale) return entry.translations # Shielded, so a cancelled caller does not cancel the load for the others. await asyncio.shield(self._start_load(locale)) @@ -138,6 +159,12 @@ def _load_blocking(self, locale: str) -> dict[str, str] | None: except Exception: return None + def _load_locked(self, locale: str) -> None: + """Load *locale* on this thread unless it is fresh, after any load of it already running on a thread.""" + with self._lock_for(locale): + if self._fresh(locale) is None: + self._store(locale, self._load_blocking(locale)) + def _reload_in_background(self, locale: str, lock: threading.Lock) -> None: try: self._store(locale, self._load_blocking(locale)) @@ -145,7 +172,7 @@ def _reload_in_background(self, locale: str, lock: threading.Lock) -> None: lock.release() def _refresh_in_background(self, locale: str) -> None: - """Reload an expired *locale* on a daemon thread, unless a reload is already running.""" + """Load *locale* on a daemon thread, unless a thread is loading it already.""" lock = self._lock_for(locale) if not lock.acquire(blocking=False): return @@ -178,8 +205,11 @@ def get_translations_blocking(self, locale: str) -> dict[str, str]: return self._store(locale, self._load_blocking(locale)) def load_all_blocking(self, locales: list[str]) -> None: - """Eagerly fetch translations for all *locales* from synchronous code.""" - _run_blocking(self.load_all(locales)) + """Eagerly fetch translations for all *locales* from synchronous code, waiting for every load.""" + stale = [loc for loc in locales if self._fresh(loc) is None] + if stale: + with ThreadPoolExecutor() as pool: + list(pool.map(self._load_locked, stale)) def get_translations_sync(self, locale: str) -> dict[str, str]: """Return the last loaded translations, even if expired, or an empty dict (never blocks).""" diff --git a/packages/gt-i18n/tests/test_translations_manager.py b/packages/gt-i18n/tests/test_translations_manager.py index 4bc6f2e..98dfbfa 100644 --- a/packages/gt-i18n/tests/test_translations_manager.py +++ b/packages/gt-i18n/tests/test_translations_manager.py @@ -6,7 +6,7 @@ from types import SimpleNamespace import pytest -from gt_i18n.i18n_manager._translations_manager import TranslationsManager +from gt_i18n.i18n_manager._translations_manager import TranslationsLoader, TranslationsManager @pytest.fixture @@ -150,6 +150,105 @@ def test_one_off_event_loops_still_refresh_after_expiry(clock: _Clock) -> None: assert calls[0] == 2 +def test_async_loader_reloads_on_the_callers_event_loop(clock: _Clock) -> None: + loops: list[asyncio.AbstractEventLoop] = [] + + async def loader(locale: str) -> dict[str, str]: + loops.append(asyncio.get_running_loop()) # a client bound to the app's loop works only there + return {"h": f"v{len(loops)}"} + + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + + async def scenario() -> asyncio.AbstractEventLoop: + await mgr.get_translations("es") + clock.now += 61 + assert await mgr.get_translations("es") == {"h": "v1"} # served at once while it reloads + await mgr.load_all(["es"]) # waits for that reload instead of starting another + return asyncio.get_running_loop() + + app_loop = asyncio.run(scenario()) + assert loops == [app_loop, app_loop] + assert mgr.get_translations_sync("es") == {"h": "v2"} + + +def test_async_reload_finishes_after_a_one_off_event_loop_closes(clock: _Clock) -> None: + finished = [0] + + async def loader(locale: str) -> dict[str, str]: + await asyncio.sleep(0.01) # still running when the caller's loop closes + finished[0] += 1 + return {"h": f"v{finished[0]}"} + + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"} + clock.now += 61 + assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"} + _settle(mgr, "es") # the reload moved to a thread when that loop closed + assert mgr.get_translations_sync("es") == {"h": "v2"} + + +def _gated(*, is_async: bool) -> tuple[TranslationsLoader, threading.Event, list[int]]: + """A loader whose second call, the first reload, waits for the returned gate; it counts its calls.""" + gate = threading.Event() + calls = [0] + + def load(locale: str) -> dict[str, str]: + calls[0] += 1 + n = calls[0] + if n == 2: + gate.wait(5) + return {"h": f"v{n}"} + + async def load_async(locale: str) -> dict[str, str]: + return load(locale) + + return (load_async if is_async else load), gate, calls + + +def test_load_all_shares_the_running_reload(clock: _Clock) -> None: + loader, gate, calls = _gated(is_async=False) + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + + async def scenario() -> None: + await mgr.get_translations("es") + clock.now += 61 + await mgr.get_translations("es") # starts the reload + load_all = asyncio.ensure_future(mgr.load_all(["es"])) + await asyncio.sleep(0.05) + gate.set() + await load_all + + asyncio.run(scenario()) + assert calls[0] == 2 # one reload, which load_all waited for + assert mgr.get_translations_sync("es") == {"h": "v2"} + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync loader", "async loader"]) +def test_load_all_blocking_shares_the_running_reload(clock: _Clock, is_async: bool) -> None: + loader, gate, calls = _gated(is_async=is_async) + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + mgr.get_translations_blocking("es") + clock.now += 61 + mgr.get_translations_blocking("es") # starts the reload + threading.Timer(0.05, gate.set).start() + mgr.load_all_blocking(["es"]) + assert calls[0] == 2 # one reload, which load_all_blocking waited for + assert mgr.get_translations_sync("es") == {"h": "v2"} + + +def test_async_read_starts_no_reload_while_a_thread_reloads(clock: _Clock) -> None: + loader, gate, calls = _gated(is_async=True) + mgr = TranslationsManager(loader, cache_expiry_time=60_000) + mgr.get_translations_blocking("es") + clock.now += 61 + mgr.get_translations_blocking("es") # reloads on a thread + assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"} + gate.set() + _settle(mgr, "es") + assert calls[0] == 2 + assert mgr.get_translations_sync("es") == {"h": "v2"} + + def test_failed_reload_keeps_previous_translations(clock: _Clock, sample_translations: dict[str, str]) -> None: calls = [0]