runtime: guard FluentLocalization._bundles against concurrent generator access

fluent.runtime/fluent/runtime/fallback.py

@@ -1,4 +1,5 @@
 import os
+import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, Any, Callable, Union, cast
 
@@ -43,6 +44,7 @@ def __init__(
         self.functions = functions
         self._bundle_cache: list[FluentBundle] = []
         self._bundle_it = self._iterate_bundles()
+        self._bundle_it_lock = threading.Lock()
 
     def format_message(
         self, msg_id: str, args: Union[dict[str, Any], None] = None
@@ -95,10 +97,14 @@ def _bundles(self) -> Generator[FluentBundle, None, None]:
         bundle_pointer = 0
         while True:
             if bundle_pointer == len(self._bundle_cache):
-                try:
-                    self._bundle_cache.append(next(self._bundle_it))
-                except StopIteration:
-                    return
+                with self._bundle_it_lock:
+                    # Re-check under the lock: another thread may have
+                    # extended the cache while we were waiting.
+                    if bundle_pointer == len(self._bundle_cache):
+                        try:
+                            self._bundle_cache.append(next(self._bundle_it))
+                        except StopIteration:
+                            return
             yield self._bundle_cache[bundle_pointer]
             bundle_pointer += 1
 

fluent.runtime/tests/test_fallback.py

@@ -78,6 +78,40 @@ def test_bundles(self, tmp_path):
         assert tuple(l10n.format_message("baz")) == ("baz in English", {})
         assert tuple(l10n.format_message("not-exists")) == ("not-exists", {})
 
+    def test_format_value_is_thread_safe(self, tmp_path):
+        # Regression test for issue #221: concurrent format_value() calls used
+        # to race on the shared _bundle_it generator and raise
+        # "ValueError: generator already executing".
+        import threading
+
+        build_file_tree(
+            tmp_path,
+            {
+                "en": {"one.ftl": "hello = world\n"},
+            },
+        )
+        l10n = FluentLocalization(
+            ["en"], ["one.ftl"], FluentResourceLoader(join(tmp_path, "{locale}"))
+        )
+        errors: list[BaseException] = []
+        start = threading.Event()
+
+        def worker():
+            start.wait()
+            try:
+                for _ in range(50):
+                    assert l10n.format_value("hello") == "world"
+            except BaseException as exc:
+                errors.append(exc)
+
+        threads = [threading.Thread(target=worker) for _ in range(8)]
+        for t in threads:
+            t.start()
+        start.set()
+        for t in threads:
+            t.join()
+        assert errors == []
+
 
 class TestResourceLoader:
     def test_all_exist(self, tmp_path):