From ac735dc48a13a743f58a6f8e0cf7283db766794d Mon Sep 17 00:00:00 2001 From: cmyui Date: Fri, 24 Jul 2026 18:52:28 -0400 Subject: [PATCH 1/3] fix: bound remote fetch connection pool acquire wait by fetch timeout The remote client acquired its pooled connection with no timeout, so when all pooled connections were busy, concurrent fetches blocked indefinitely in pool.acquire() -- a wait that fetch_timeout_millis never covered, since it only bounds the socket read after acquisition. Pass fetch_timeout_millis as the acquire timeout and map the pool's existing EmptyPoolError to FetchException(408). 408 lands in the non-retryable bucket of __should_retry_fetch by design: retrying a starved pool would re-enter the same queue and add pressure. Co-Authored-By: Claude Fable 5 --- src/amplitude_experiment/remote/client.py | 8 ++++++-- src/amplitude_experiment/remote/config.py | 3 ++- tests/remote/client_test.py | 24 ++++++++++++++++++++++- 3 files changed, 31 insertions(+), 4 deletions(-) diff --git a/src/amplitude_experiment/remote/client.py b/src/amplitude_experiment/remote/client.py index 8db2653..e6203e6 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 FetchException(408, 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..dd27b5b 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,27 @@ 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_with_fetch_exception(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(FetchException) as ctx: + client._RemoteEvaluationClient__do_fetch(User(user_id='test_user')) + elapsed = time.time() - start + self.assertEqual(408, ctx.exception.status_code) + # 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_is_not_retried(self): + # Retrying a starved pool would re-enter the same queue; 408 must be non-retryable. + should_retry = RemoteEvaluationClient._RemoteEvaluationClient__should_retry_fetch + self.assertFalse(should_retry(FetchException(408, "pool acquire timed out"))) + 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 +75,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': { From 207ae5a30089f5d4e29ffd760a9d68aa2ce243db Mon Sep 17 00:00:00 2001 From: cmyui Date: Fri, 24 Jul 2026 19:00:02 -0400 Subject: [PATCH 2/3] fix: classify pool acquire timeout like a read timeout Raising FetchException(408) put the pool-wait timeout in the non-retryable 4xx bucket, where __fetch_internal swallows the error and fetch_v2 returns None with no signal (flagged by Bugbot). Raise a builtin TimeoutError instead -- the same classification as the socket read timeout -- so both violations of fetch_timeout_millis behave identically: empty variants at fetch_retries=0, bounded retries then raise when retries are configured, and async callbacks receive the error. Adds an end-to-end parity test on fetch_v2. Co-Authored-By: Claude Fable 5 --- src/amplitude_experiment/remote/client.py | 4 ++-- tests/remote/client_test.py | 23 +++++++++++++++++------ 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/src/amplitude_experiment/remote/client.py b/src/amplitude_experiment/remote/client.py index e6203e6..b4f4995 100644 --- a/src/amplitude_experiment/remote/client.py +++ b/src/amplitude_experiment/remote/client.py @@ -152,8 +152,8 @@ def __do_fetch(self, user, fetch_options: FetchOptions = None): try: conn = self._connection_pool.acquire(timeout=self.config.fetch_timeout_millis / 1000) except EmptyPoolError: - raise FetchException(408, f"Timed out waiting {self.config.fetch_timeout_millis}ms for a connection " - f"from the pool (max_size={self._connection_pool.max_size})") + 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/tests/remote/client_test.py b/tests/remote/client_test.py index dd27b5b..573294f 100644 --- a/tests/remote/client_test.py +++ b/tests/remote/client_test.py @@ -42,26 +42,37 @@ 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_with_fetch_exception(self): + 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(FetchException) as ctx: + with self.assertRaises(TimeoutError): client._RemoteEvaluationClient__do_fetch(User(user_id='test_user')) elapsed = time.time() - start - self.assertEqual(408, ctx.exception.status_code) # 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_is_not_retried(self): - # Retrying a starved pool would re-enter the same queue; 408 must be non-retryable. + 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.assertFalse(should_retry(FetchException(408, "pool acquire timed out"))) + 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, From fd125084a525b355be7362df5d214a0df041b9af Mon Sep 17 00:00:00 2001 From: cmyui Date: Tue, 28 Jul 2026 15:35:57 -0400 Subject: [PATCH 3/3] fix: make pool acquire timeout opt-in via fetch_pool_acquire_timeout_millis Default None preserves today's unbounded acquire wait exactly; setting the new config bounds it, surfacing exhaustion as TimeoutError with the same retry classification as a socket read timeout. Reworked from reusing fetch_timeout_millis so the PR carries no default-behavior change; whether to bound by default is posed to maintainers in the PR description instead. Co-Authored-By: Claude Fable 5 --- src/amplitude_experiment/remote/client.py | 7 +++++-- src/amplitude_experiment/remote/config.py | 9 +++++++-- tests/remote/client_test.py | 11 +++++++++-- 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/src/amplitude_experiment/remote/client.py b/src/amplitude_experiment/remote/client.py index b4f4995..a293edd 100644 --- a/src/amplitude_experiment/remote/client.py +++ b/src/amplitude_experiment/remote/client.py @@ -149,10 +149,13 @@ def __do_fetch(self, user, fetch_options: FetchOptions = None): json.dumps(fetch_options.flagKeys, separators=(",", ":")).encode("utf-8") ).rstrip(b"=").decode("utf-8") + acquire_timeout_millis = self.config.fetch_pool_acquire_timeout_millis try: - conn = self._connection_pool.acquire(timeout=self.config.fetch_timeout_millis / 1000) + conn = self._connection_pool.acquire( + timeout=acquire_timeout_millis / 1000 if acquire_timeout_millis is not None else None + ) except EmptyPoolError: - raise TimeoutError(f"Timed out waiting {self.config.fetch_timeout_millis}ms for a connection " + raise TimeoutError(f"Timed out waiting {acquire_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: diff --git a/src/amplitude_experiment/remote/config.py b/src/amplitude_experiment/remote/config.py index d9cb4fa..900202a 100644 --- a/src/amplitude_experiment/remote/config.py +++ b/src/amplitude_experiment/remote/config.py @@ -17,6 +17,7 @@ def __init__(self, debug=False, fetch_retry_backoff_max_millis=10000, fetch_retry_backoff_scalar=1.5, fetch_retry_timeout_millis=10000, + fetch_pool_acquire_timeout_millis=None, server_zone: ServerZone = ServerZone.US, logger=None): """ @@ -25,8 +26,7 @@ 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(). 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. + triggered by calling start() or setUser(). 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. @@ -34,6 +34,10 @@ def __init__(self, debug=False, greater than the max, the max is used for all subsequent retries. fetch_retry_backoff_scalar (float): Scales the minimum backoff exponentially. fetch_retry_timeout_millis (int): The request timeout for retrying fetch requests. + fetch_pool_acquire_timeout_millis (int | None): Maximum time, in milliseconds, a fetch may wait to + acquire a connection from the connection pool when all pooled connections are busy with concurrent + fetches. None (the default) preserves the existing behavior of waiting indefinitely. When set, an + exhausted wait raises TimeoutError, classified for retries exactly like a socket read timeout. server_zone (str): Select the Amplitude data center to get flags and variants from, US or EU. logger (logging.Logger): Optional logger instance. If provided, this logger will be used instead of creating a new one. The debug flag only applies when no logger is provided. @@ -49,6 +53,7 @@ def __init__(self, debug=False, self.fetch_retry_backoff_max_millis = fetch_retry_backoff_max_millis self.fetch_retry_backoff_scalar = fetch_retry_backoff_scalar self.fetch_retry_timeout_millis = fetch_retry_timeout_millis + self.fetch_pool_acquire_timeout_millis = fetch_pool_acquire_timeout_millis self.server_zone = server_zone if server_url == DEFAULT_SERVER_URL and server_zone == ServerZone.EU: self.server_url = EU_SERVER_URL diff --git a/tests/remote/client_test.py b/tests/remote/client_test.py index 573294f..f9871ec 100644 --- a/tests/remote/client_test.py +++ b/tests/remote/client_test.py @@ -42,8 +42,13 @@ def test_fetch_async(self): user = User(user_id='test_user') self.client.fetch_async(user, self.callback_for_async) + def test_pool_acquire_wait_unbounded_by_default(self): + assert RemoteEvaluationConfig().fetch_pool_acquire_timeout_millis is None + def test_fetch_pool_exhausted_times_out(self): - client = RemoteEvaluationClient(API_KEY, RemoteEvaluationConfig(fetch_timeout_millis=100)) + client = RemoteEvaluationClient( + API_KEY, RemoteEvaluationConfig(fetch_pool_acquire_timeout_millis=100) + ) # Occupy the pool's only connection so the next fetch must wait for it. held = client._connection_pool.acquire() try: @@ -65,7 +70,9 @@ def test_pool_acquire_timeout_classified_like_read_timeout(self): 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)) + client = RemoteEvaluationClient( + API_KEY, RemoteEvaluationConfig(fetch_pool_acquire_timeout_millis=100) + ) held = client._connection_pool.acquire() try: variants = client.fetch_v2(User(user_id='test_user'))