fix: bound remote fetch connection pool acquire wait by fetch timeout

src/amplitude_experiment/remote/client.py

@@ -7,7 +7,7 @@
 
 from .config import RemoteEvaluationConfig
 from .fetch_options import FetchOptions
-from ..connection_pool import HTTPConnectionPool
+from ..connection_pool import EmptyPoolError, HTTPConnectionPool
 from ..exception import FetchException
 from ..user import User
 from ..util.deprecated import deprecated
@@ -149,7 +149,11 @@ def __do_fetch(self, user, fetch_options: FetchOptions = None):
                 json.dumps(fetch_options.flagKeys, separators=(",", ":")).encode("utf-8")
             ).rstrip(b"=").decode("utf-8")
 
-        conn = self._connection_pool.acquire()
+        try:
+            conn = self._connection_pool.acquire(timeout=self.config.fetch_timeout_millis / 1000)
+        except EmptyPoolError:
+            raise TimeoutError(f"Timed out waiting {self.config.fetch_timeout_millis}ms for a connection "
+                               f"from the pool (max_size={self._connection_pool.max_size})")
         body = user_context.to_json().encode('utf8')
         if len(body) > 8000:
             self.logger.warning(f"[Experiment] encoded user object length ${len(body)} "

src/amplitude_experiment/remote/config.py

@@ -25,7 +25,8 @@ def __init__(self, debug=False,
                 debug (bool): Set to true to log some extra information to the console.
                 server_url (str): The server endpoint from which to request variants.
                 fetch_timeout_millis (int): The request timeout, in milliseconds, used when fetching variants
-                  triggered by calling start() or setUser().
+                  triggered by calling start() or setUser(). Also bounds how long a fetch may wait to acquire a
+                  connection from the connection pool when all connections are in use by concurrent fetches.
                 fetch_retries (int): The number of retries to attempt before failing.
                 fetch_retry_backoff_min_millis (int): Retry backoff minimum (starting backoff delay) in milliseconds.
                   The minimum backoff is scaled by `fetch_retry_backoff_scalar` after each retry failure.

tests/remote/client_test.py

@@ -1,4 +1,5 @@
 import json
+import time
 import unittest
 from unittest import mock
 
@@ -41,6 +42,38 @@ def test_fetch_async(self):
         user = User(user_id='test_user')
         self.client.fetch_async(user, self.callback_for_async)
 
+    def test_fetch_pool_exhausted_times_out(self):
+        client = RemoteEvaluationClient(API_KEY, RemoteEvaluationConfig(fetch_timeout_millis=100))
+        # Occupy the pool's only connection so the next fetch must wait for it.
+        held = client._connection_pool.acquire()
+        try:
+            start = time.time()
+            with self.assertRaises(TimeoutError):
+                client._RemoteEvaluationClient__do_fetch(User(user_id='test_user'))
+            elapsed = time.time() - start
+            # Must fail promptly (bounded by fetch_timeout_millis), not hang indefinitely.
+            self.assertLess(elapsed, 5)
+        finally:
+            client._connection_pool.release(held)
+            client.close()
+
+    def test_pool_acquire_timeout_classified_like_read_timeout(self):
+        # A pool-wait timeout must take the same retry path as a socket read timeout.
+        should_retry = RemoteEvaluationClient._RemoteEvaluationClient__should_retry_fetch
+        self.assertTrue(should_retry(TimeoutError("pool acquire timed out")))
+
+    def test_fetch_v2_pool_exhausted_matches_read_timeout_behavior(self):
+        # With fetch_retries=0, a pool-wait timeout must surface exactly like a read
+        # timeout: an empty variants dict, not None and not an exception.
+        client = RemoteEvaluationClient(API_KEY, RemoteEvaluationConfig(fetch_timeout_millis=100))
+        held = client._connection_pool.acquire()
+        try:
+            variants = client.fetch_v2(User(user_id='test_user'))
+            self.assertEqual({}, variants)
+        finally:
+            client._connection_pool.release(held)
+            client.close()
+
     def test_fetch_failed_with_retry(self):
         with RemoteEvaluationClient(API_KEY, RemoteEvaluationConfig(debug=False, fetch_retries=1,
                                                                     fetch_timeout_millis=1)) as client:
@@ -53,7 +86,7 @@ def test_fetch_with_fetch_options(self):
             user = User(user_id='test_user')
 
             mock_conn = mock.MagicMock()
-            client._connection_pool.acquire = lambda: mock_conn
+            client._connection_pool.acquire = lambda **kwargs: mock_conn
             mock_conn.request.return_value = mock.MagicMock(status=200)
             mock_conn.request.return_value.read.return_value = json.dumps({
                 'sdk-ci-test': {