diff --git a/src/amplitude_experiment/remote/client.py b/src/amplitude_experiment/remote/client.py index 8db2653..b4f4995 100644 --- a/src/amplitude_experiment/remote/client.py +++ b/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)} " diff --git a/src/amplitude_experiment/remote/config.py b/src/amplitude_experiment/remote/config.py index c4efdcc..d9cb4fa 100644 --- a/src/amplitude_experiment/remote/config.py +++ b/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. diff --git a/tests/remote/client_test.py b/tests/remote/client_test.py index 45c45b8..573294f 100644 --- a/tests/remote/client_test.py +++ b/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': {