From ac735dc48a13a743f58a6f8e0cf7283db766794d Mon Sep 17 00:00:00 2001 From: cmyui Date: Fri, 24 Jul 2026 18:52:28 -0400 Subject: [PATCH 1/2] 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/2] 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,